HPCC-9609 - Tidy up dataLinkStart calls

Unecessary calls passing in activity string/id, when CThorDataLink
already had activity with same info.

Also resolve a uninitialized warning picked up by Valgrind.

Signed-off-by: Jake Smith <jake.smith@lexisnexis.com>

thorlcr/activities/aggregate/thaggregateslave.cpp

@@ -157,7 +157,7 @@ public:
         ActivityTimer s(totalCycles, timeActivities, NULL);
         doStart();
         eof = false;
-        dataLinkStart("AGGREGATE", container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {
@@ -270,7 +270,7 @@ public:
         aggrowif.setown(createRowInterfaces(helper->queryAggregateRecordSize(),queryActivityId(),queryCodeContext()));
         partResult.setAllocator(aggrowif->queryRowAllocator()).ensureRow();
         helper->clearAggregate(partResult);
-        dataLinkStart("THROUGHAGGREGATE", container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {

thorlcr/activities/aggregate/thgroupaggregateslave.cpp

@@ -47,7 +47,7 @@ public:
         input=inputs.item(0);
         startInput(input);
         ungroupedExistsAggregate = (container.getKind() == TAKexistsaggregate) && !input->isGrouped();
-        dataLinkStart("GROUPAGGREGATE", container.queryId());
+        dataLinkStart();
     }
 
     void stop()

thorlcr/activities/catch/thcatchslave.cpp

@@ -27,7 +27,6 @@ class CCatchSlaveActivityBase : public CSlaveActivity, public CThorDataLink
 protected:
     Owned<IThorDataLink> input;
     IHThorCatchArg *helper;
-    StringBuffer kindStr;
     bool eos;
 
 public:
@@ -39,8 +38,6 @@ public:
     virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
     {
         helper = static_cast <IHThorCatchArg *> (queryHelper());
-        kindStr.append(activityKindStr(container.getKind()));
-        kindStr.toUpperCase();
         eos = false;
         appendOutputLinked(this);
     }
@@ -49,7 +46,7 @@ public:
         input.set(inputs.item(0));
         startInput(input);
         eos = false;
-        dataLinkStart(kindStr.str(), container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {

thorlcr/activities/choosesets/thchoosesetsslave.cpp

@@ -73,7 +73,7 @@ public:
         ActPrintLog("CHOOSESETS: Is Local");
         input.set(inputs.item(0));
         startInput(input);
-        dataLinkStart("CHOOSESETS", container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {
@@ -166,7 +166,7 @@ public:
         done = false;
         input.setown(createDataLinkSmartBuffer(this, inputs.item(0),CHOOSESETS_SMART_BUFFER_SIZE,isSmartBufferSpillNeeded(this),false,RCUNBOUND,NULL,false,&container.queryJob().queryIDiskUsage())); // only allow spill if input can stall
         startInput(input);
-        dataLinkStart("CHOOSESETS", container.queryId());       
+        dataLinkStart();
     }
     virtual void stop()
     {
@@ -320,7 +320,7 @@ public:
         first = true;
         input.setown(createDataLinkSmartBuffer(this, inputCounter,CHOOSESETSPLUS_SMART_BUFFER_SIZE,true,false,RCUNBOUND,this,false,&container.queryJob().queryIDiskUsage())); // read all input
         startInput(input);
-        dataLinkStart("CHOOSESETS", container.queryId());       
+        dataLinkStart();
     }
     virtual void stop()
     {

thorlcr/activities/countproject/thcountprojectslave.cpp

@@ -29,7 +29,7 @@ protected:
     void start()
     {
         count = 0;
-        dataLinkStart("COUNTPROJECT", container.queryId());
+        dataLinkStart();
     }
 public:
     IMPLEMENT_IINTERFACE_USING(CSimpleInterface);

thorlcr/activities/csvread/thcsvrslave.cpp

@@ -410,7 +410,7 @@ public:
             }
         }
         out.setown(createSequentialPartHandler(partHandler, partDescs, false));
-        dataLinkStart("CCsvReadSlaveActivity", container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {

thorlcr/activities/degroup/thdegroupslave.cpp

@@ -44,7 +44,7 @@ public:
 
         startInput(input);
         if(!input->isGrouped()) ActPrintLog("DEGROUP: Degrouping non-grouped input!");
-        dataLinkStart("DEGROUP", container.queryId());
+        dataLinkStart();
     }
     void stop()
     {

thorlcr/activities/diskread/thdiskreadslave.cpp

@@ -452,7 +452,7 @@ public:
         ActivityTimer s(totalCycles, timeActivities, NULL);
         CDiskReadSlaveActivityRecord::start();
         out = createSequentialPartHandler(partHandler, partDescs, grouped); // **
-        dataLinkStart("DISKREAD", container.queryId());
+        dataLinkStart();
     }
     virtual bool isGrouped() { return grouped; }
 
@@ -608,7 +608,7 @@ public:
         ActivityTimer s(totalCycles, timeActivities, NULL);
         CDiskReadSlaveActivityRecord::start();
         out = createSequentialPartHandler(partHandler, partDescs, false);
-        dataLinkStart("DISKNORMALIZE", container.queryId());
+        dataLinkStart();
     }
     virtual bool isGrouped() { return false; }
 
@@ -732,7 +732,7 @@ public:
         ActivityTimer s(totalCycles, timeActivities, NULL);
         CDiskReadSlaveActivityRecord::start();
         eoi = hadElement = false;
-        dataLinkStart("DISKAGGREGATE", container.queryId());
+        dataLinkStart();
     }
 
 // IRowStream
@@ -855,7 +855,7 @@ public:
             totalCountKnown = true;
             preknownTotalCount = 0;
         }
-        dataLinkStart("DISKCOUNT", container.queryId());
+        dataLinkStart();
     }
 
 // IRowStream
@@ -967,7 +967,7 @@ public:
         gathered = eoi = false;
         localAggTable.setown(new CThorRowAggregator(*this, *helper, *helper));
         localAggTable->start(queryRowAllocator());
-        dataLinkStart("DISKGROUPAGGREGATE", container.queryId());
+        dataLinkStart();
     }
     virtual void getMetaInfo(ThorDataLinkMetaInfo &info)
     {

thorlcr/activities/enth/thenthslave.cpp

@@ -89,7 +89,7 @@ public:
         numerator = validRC(helper->getProportionNumerator());
         input.set(inputs.item(0));
         startInput(input);
-        dataLinkStart("ENTH", container.queryId());
+        dataLinkStart();
 
         if (localCountReq)
         {

thorlcr/activities/fetch/thfetchslave.cpp

@@ -376,7 +376,7 @@ public:
         };
 
         startInput(inputs.item(0));
-        dataLinkStart("FETCH", container.queryId());
+        dataLinkStart();
 
         IThorDataLink *in = inputs.item(0);
         Owned<IRowInterfaces> keyInIf;

thorlcr/activities/filter/thfilterslave.cpp

@@ -35,12 +35,12 @@ public:
     {
         appendOutputLinked(this);
     }
-    void start(const char *act)
+    void start()
     {   
         input = inputs.item(0);
         anyThisGroup = false;
         startInput(input);
-        dataLinkStart(act, container.queryId());
+        dataLinkStart();
     }
     void stop()
     {
@@ -80,7 +80,7 @@ public:
         ActivityTimer s(totalCycles, timeActivities, NULL);
         matched = 0;
         abortSoon = !helper->canMatchAny();
-        CFilterSlaveActivityBase::start("FILTER");
+        CFilterSlaveActivityBase::start();
     }
     CATCH_NEXTROW()
     {
@@ -189,7 +189,7 @@ public:
         ActivityTimer s(totalCycles, timeActivities, NULL);
         abortSoon = !helper->canMatchAny();
         recordCount = 0;
-        CFilterSlaveActivityBase::start("FILTERPROJECT");
+        CFilterSlaveActivityBase::start();
     }
     CATCH_NEXTROW()
     {
@@ -259,7 +259,7 @@ public:
     {   
         ActivityTimer s(totalCycles, timeActivities, NULL);
         abortSoon = !helper->canMatchAny();
-        CFilterSlaveActivityBase::start("FILTERGROUP");
+        CFilterSlaveActivityBase::start();
     }
     CATCH_NEXTROW()
     {

thorlcr/activities/firstn/thfirstnslave.cpp

@@ -60,7 +60,7 @@ public:
         input.set(inputs.item(0));
         startInput(input);
         stopped = false;
-        dataLinkStart("FIRSTN", container.queryId());
+        dataLinkStart();
     }
     void stop()
     {

thorlcr/activities/funnel/thfunnelslave.cpp

@@ -337,7 +337,7 @@ public:
                 {
                     ActivityTimer s(activity.getTotalCyclesRef(), activity.queryTimeActivities(), NULL);
                     out.setown(createParallelFunnel(activity, inputs.getArray(), inputs.ordinality(), true, abortSoon));
-                    dataLinkStart("FUNNEL", activity.queryContainer().queryId());
+                    dataLinkStart();
                 }
                 const void *nextRow()
                 {
@@ -399,8 +399,7 @@ public:
             }
             if (!current) current = input;
         }
-
-        dataLinkStart("FUNNEL", container.queryId());
+        dataLinkStart();
     }
 
     void stop()
@@ -561,8 +560,7 @@ public:
                 throw;
             }
         }
-
-        dataLinkStart("COMBINE", container.queryId());
+        dataLinkStart();
     }
     void stop()
     {
@@ -666,7 +664,7 @@ public:
                 throw;
             }
         }
-        dataLinkStart("REGROUP", container.queryId());
+        dataLinkStart();
     }
     void stop()
     {
@@ -786,7 +784,7 @@ public:
             }
         }
         eoi = 0 == inputs.ordinality();
-        dataLinkStart("NONEMPTY", container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {

thorlcr/activities/group/thgroupslave.cpp

@@ -82,7 +82,7 @@ public:
         input = inputs.item(0);
         stream.set(input);
         startInput(input);
-        dataLinkStart("GROUP", container.queryId());        
+        dataLinkStart();
 
         next.setown(getNext());
 

thorlcr/activities/hashdistrib/thhashdistribslave.cpp

@@ -1722,7 +1722,7 @@ public:
             Owned<IRowInterfaces> myRowIf = getRowInterfaces(); // avoiding circular link issues
             out.setown(distributor->connect(myRowIf, instrm, ihash, mergecmp));
         }
-        dataLinkStart("HASHDISTRIB", container.queryId());
+        dataLinkStart();
     }
     void start()
     {
@@ -2475,7 +2475,6 @@ protected:
     IRowStream *initialInput;
     Owned<IRowStream> currentInput;
     bool inputstopped, eos, lastEog, extractKey, local, isVariable, grouped;
-    const char *actTxt;
     IHThorHashDedupArg *helper;
     IHash *iHash, *iKeyHash;
     ICompare *iCompare, *rowKeyCompare;
@@ -2526,7 +2525,6 @@ public:
         : CSlaveActivity(_container), CThorDataLink(this), local(_local)
     {
         input = initialInput = NULL;
-        actTxt = NULL;
         initialNumBuckets = 0;
         inputstopped = eos = lastEog = extractKey = local = isVariable = grouped = false;
         helper = NULL;
@@ -2604,7 +2602,7 @@ public:
         }
         ensureNumHashTables(initialNumBuckets);
         bucketHandler->init(initialNumBuckets);
-        dataLinkStart(actTxt, container.queryId());
+        dataLinkStart();
     }
     void stopInput()
     {
@@ -2617,7 +2615,7 @@ public:
     }
     void kill()
     {
-        ActPrintLog("%s: kill", actTxt);
+        ActPrintLog("kill");
         currentInput.clear();
         bucketHandler.clear();
         bucketHandlerStack.kill();
@@ -3104,11 +3102,10 @@ public:
     LocalHashDedupSlaveActivity(CGraphElementBase *container)
         : HashDedupSlaveActivityBase(container, true)
     {
-        actTxt = "LOCALHASHDEDUP";
     }
     void stop()
     {
-        ActPrintLog("%s: stopping", actTxt);
+        ActPrintLog("stopping");
         stopInput();
         dataLinkStop();
     }
@@ -3132,7 +3129,6 @@ public:
     GlobalHashDedupSlaveActivity(CGraphElementBase *container)
         : HashDedupSlaveActivityBase(container, false)
     {
-        actTxt = "HASHDEDUP";
         distributor = NULL;
         mptag = TAG_NULL;
     }
@@ -3172,7 +3168,7 @@ public:
 
     void stop()
     {
-        ActPrintLog("%s: stopping", actTxt);
+        ActPrintLog("stopping");
         if (instrm)
         {
             instrm->stop();
@@ -3299,7 +3295,7 @@ public:
             }
         }
         joinhelper->init(strmL, strmR, ::queryRowAllocator(inL), ::queryRowAllocator(inR), ::queryRowMetaData(inL), &abortSoon);
-        dataLinkStart("HASHJOIN", container.queryId());
+        dataLinkStart();
     }
     void stopInput()
     {
@@ -3570,7 +3566,7 @@ public:
             ActPrintLog("Table after distribution contains %d entries", localAggTable->elementCount());
         }
         eos = false;
-        dataLinkStart("HASHAGGREGATE", container.queryId());
+        dataLinkStart();
     }
     void stop()
     {

thorlcr/activities/indexread/thindexreadslave.cpp

@@ -662,7 +662,7 @@ public:
             keyedLimitCount = 0;            
         else
             eoi = true; // otherwise delayed until calc. in nextRow()
-        dataLinkStart("INDEXREAD", container.queryId());
+        dataLinkStart();
     }
     virtual void getMetaInfo(ThorDataLinkMetaInfo &info)
     {
@@ -816,7 +816,7 @@ public:
         localAggTable.setown(new CThorRowAggregator(*this, *helper, *helper));
         localAggTable->start(queryRowAllocator());
         gathered = eoi = false;
-        dataLinkStart("INDEXGROUPAGGREGATE", container.queryId());
+        dataLinkStart();
     }
 // IRowStream
     virtual void stop()
@@ -921,7 +921,7 @@ public:
             totalCountKnown = true;
             preknownTotalCount = 0;
         }
-        dataLinkStart("INDEXCOUNT", container.queryId());
+        dataLinkStart();
     }
 
 // IRowStream
@@ -1106,7 +1106,7 @@ public:
         }
         else
             eoi = true;
-        dataLinkStart("INDEXNORMALIZE", container.queryId());
+        dataLinkStart();
     }
 
 // IRowStream
@@ -1268,7 +1268,7 @@ public:
         ActivityTimer s(totalCycles, timeActivities, NULL);
         eoi = hadElement = false;
         partn = 0;
-        dataLinkStart("INDEXAGGREGATE", container.queryId());
+        dataLinkStart();
     }
 
 // IRowStream

thorlcr/activities/iterate/thgroupiterateslave.cpp

@@ -55,7 +55,7 @@ public:
         count = 0;
         input = inputs.item(0);
         startInput(input);
-        dataLinkStart("GROUPITERATE", container.queryId());
+        dataLinkStart();
         RtlDynamicRowBuilder r(queryRowAllocator());
         size32_t sz = helper->createDefault(r);
         defaultLeft.setown(r.finalizeRowClear(sz));
@@ -153,7 +153,7 @@ public:
         eogNext = false;    
         input = inputs.item(0);
         startInput(input);
-        dataLinkStart("GROUPPROCESS", container.queryId());
+        dataLinkStart();
     }
     void stop()
     {

thorlcr/activities/iterate/thiterateslave.cpp

@@ -92,8 +92,7 @@ public:
             ActPrintLog(e,"ITERATE");
             throw;
         }
-        dataLinkStart("ITERATOR", container.queryId());
-
+        dataLinkStart();
     }
     void stop()
     {
@@ -303,7 +302,7 @@ public:
         ActivityTimer s(totalCycles, timeActivities, NULL);
         eof = !container.queryLocalOrGrouped() && !firstNode();
         count = 0;
-        dataLinkStart("CHILDITERATOR", container.queryId());
+        dataLinkStart();
     }
     void stop()
     {
@@ -359,7 +358,7 @@ public:
     void start()
     {
         ActivityTimer s(totalCycles, timeActivities, NULL);
-        dataLinkStart("LINKEDRAWITERATOR", container.queryId());
+        dataLinkStart();
         dohere = container.queryLocalOrGrouped() || firstNode();
     }
     void stop()
@@ -423,7 +422,7 @@ public:
         eof = isLocal ? false : !firstNode();
         if (!eof)
             rows.setown(helper->createInput());
-        dataLinkStart("STREAMEDITERATOR", container.queryId());
+        dataLinkStart();
     }
     void stop()
     {

thorlcr/activities/join/thjoinslave.cpp

@@ -69,7 +69,6 @@ class JoinSlaveActivity : public CSlaveActivity, public CThorDataLink, implement
     bool islocal;
     Owned<IBarrier> barrier;
     SocketEndpoint server;
-    StringBuffer activityName;
 
 #ifdef _TESTING
     bool started;
@@ -230,7 +229,7 @@ public:
 
     void doDataLinkStart(bool denorm)
     {
-        dataLinkStart(activityName, container.queryId());
+        dataLinkStart();
         CriticalBlock b(joinHelperCrit);
         if (denorm)
             joinhelper.setown(createDenormalizeHelper(*this, helperdn, queryRowAllocator()));
@@ -434,7 +433,7 @@ public:
             stopInput1();
         }
         if (isemptylhs&&((helper->getJoinFlags()&JFrightouter)==0)) {
-            ActPrintLog("%s: ignoring RHS as LHS empty", activityName.str());
+            ActPrintLog("ignoring RHS as LHS empty");
             strm2.setown(createNullRowStream());
             stopInput2();
         }

thorlcr/activities/keyedjoin/thkeyedjoinslave.cpp

@@ -2037,7 +2037,7 @@ public:
         pool->setOrdering(preserveGroups, preserveOrder);
         resultDistStream->setInput(input);
 
-        dataLinkStart("KEYEDJOIN", container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {

thorlcr/activities/limit/thlimitslave.cpp

@@ -30,21 +30,20 @@ protected:
     bool eos, eogNext, stopped, resultSent, anyThisGroup;
     IThorDataLink *input;
     IHThorLimitArg *helper;
-    StringAttr msg;
 
     void stopInput(rowcount_t c)
     {
         if (!stopped) {
             stopped = true;
             sendResult(c);
-            CSlaveActivity::stopInput(input, msg);
+            CSlaveActivity::stopInput(input);
         }
     }
 
 public:
     IMPLEMENT_IINTERFACE_USING(CSimpleInterface);
 
-    CLimitSlaveActivityBase(CGraphElementBase *_container, const char *_msg) : CSlaveActivity(_container), msg(_msg), CThorDataLink(this)
+    CLimitSlaveActivityBase(CGraphElementBase *_container) : CSlaveActivity(_container), CThorDataLink(this)
     {
         input = NULL;       
     }
@@ -65,7 +64,7 @@ public:
         anyThisGroup = eogNext = false;
         startInput(input);
         rowLimit = (rowcount_t)helper->getRowLimit();
-        dataLinkStart(msg, container.queryId());
+        dataLinkStart();
     }
     void sendResult(rowcount_t r)
     {
@@ -95,7 +94,7 @@ class CLimitSlaveActivity : public CLimitSlaveActivityBase, public CThorSteppabl
 {
 public:
 
-    CLimitSlaveActivity(CGraphElementBase *container) : CLimitSlaveActivityBase(container, "LIMIT"), CThorSteppable(this) { }
+    CLimitSlaveActivity(CGraphElementBase *container) : CLimitSlaveActivityBase(container), CThorSteppable(this) { }
     
     CATCH_NEXTROW()
     {
@@ -229,7 +228,7 @@ class CSkipLimitSlaveActivity : public CLimitSlaveActivityBase
     }
 
 public:
-    CSkipLimitSlaveActivity(CGraphElementBase *container,bool _rowTransform) : CLimitSlaveActivityBase(container, "SKIPLIMIT")
+    CSkipLimitSlaveActivity(CGraphElementBase *container,bool _rowTransform) : CLimitSlaveActivityBase(container)
     {
         ptrIndex = 0;
         limitChecked = eof = false;

thorlcr/activities/lookupjoin/thlookupjoinslave.cpp

@@ -35,7 +35,6 @@
 
 enum join_t { JT_Undefined, JT_Inner, JT_LeftOuter, JT_RightOuter, JT_LeftOnly, JT_RightOnly, JT_LeftOnlyTransform };
 enum joinkind_t { join_lookup, join_all, denormalize_lookup, denormalize_all };
-const char *joinActName[4] = { "LOOKUPJOIN", "ALLJOIN", "LOOKUPDENORMALIZE", "ALLDENORMALIZE" };
 
 
 #define MAX_SEND_SIZE 0x100000 // 1MB
@@ -609,7 +608,6 @@ class CLookupJoinActivity : public CSlaveActivity, public CThorDataLink, impleme
     }
 protected:
     joinkind_t joinKind;
-    StringAttr joinStr;
 public:
     IMPLEMENT_IINTERFACE_USING(CSimpleInterface);
 
@@ -647,10 +645,7 @@ public:
 // IThorSlaveActivity overloaded methods
     virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
     {
-        StringBuffer js(joinActName[(int)joinKind]);
-        js.append("(").append(container.queryId()).append(")");
-        joinStr.set(js.str());
-        ActPrintLog("%s: init",joinStr.get());
+        ActPrintLog("init");
         appendOutputLinked(this);
 
         eos = false;
@@ -736,7 +731,7 @@ public:
         while (slaves--)
             rhsNodeRows.append(new CThorExpandingRowArray(*this, NULL, true)); // true, nulls not needed?
         StringBuffer str;
-        ActPrintLog("%s: Join type is %s", joinStr.get(), getJoinTypeStr(str).str());
+        ActPrintLog("Join type is %s", getJoinTypeStr(str).str());
     }
     virtual void onInputStarted(IException *except)
     {
@@ -749,7 +744,7 @@ public:
     }
     virtual void onInputFinished(rowcount_t count)
     {
-        ActPrintLog("%s: LHS input finished, %"RCPF"d rows read", joinStr.get(), count);
+        ActPrintLog("LHS input finished, %"RCPF"d rows read", count);
     }
     virtual void start()
     {
@@ -768,7 +763,6 @@ public:
         leftAllocator.set(::queryRowAllocator(left));
         outputMeta.set(left->queryFromActivity()->queryContainer().queryHelper()->queryOutputMeta());
         left.setown(createDataLinkSmartBuffer(this,left,LOOKUPJOINL_SMART_BUFFER_SIZE,isSmartBufferSpillNeeded(left->queryFromActivity()),grouped,RCUNBOUND,this,false,&container.queryJob().queryIDiskUsage()));       
-        StringBuffer str(joinStr);
         startInput(left);
         right.set(inputs.item(1));
         rightAllocator.set(::queryRowAllocator(right));
@@ -825,7 +819,7 @@ public:
             defaultRight.setown(rr.finalizeRowClear(rrsz));
         if (rlsz)
             defaultLeft.setown(rl.finalizeRowClear(rlsz));
-        dataLinkStart(joinStr, container.queryId());
+        dataLinkStart();
     }
     virtual void abort()
     {
@@ -1328,42 +1322,26 @@ public:
                 rhs.ensure((rowidx_t)rhsTotalCount);
             }
         }
-        Owned<IException> exception;
-        try
+        if (needGlobal)
         {
-            if (needGlobal)
-            {
-                rowProcessor.start();
-                broadcaster.start(this, mpTag, stopping);
-                sendRHS();
-                broadcaster.end();
-                rowProcessor.wait();
-            }
-            else if (!stopping)
-            {
-                while (!abortSoon)
-                {
-                    OwnedConstThorRow row = right->ungroupedNextRow();
-                    if (!row)
-                        break;
-                    addRow(row.getClear());
-                }
-            }
-            if (!stopping)
-                prepareRHS();
+            rowProcessor.start();
+            broadcaster.start(this, mpTag, stopping);
+            sendRHS();
+            broadcaster.end();
+            rowProcessor.wait();
         }
-        catch (IOutOfMemException *e) { exception.setown(e); }
-        if (exception.get())
+        else if (!stopping)
         {
-            StringBuffer errStr(joinStr);
-            errStr.append("(").append(container.queryId()).appendf(") right-hand side is too large (%"I64F"u bytes in %"RIPF"d rows) for %s : (",(unsigned __int64) rhs.serializedSize(),rhs.ordinality(),joinStr.get());
-            errStr.append(exception->errorCode()).append(", ");
-            exception->errorMessage(errStr);
-            errStr.append(")");
-            IException *e2 = MakeActivityException(this, TE_TooMuchData, "%s", errStr.str());
-            ActPrintLog(e2);
-            throw e2;
+            while (!abortSoon)
+            {
+                OwnedConstThorRow row = right->ungroupedNextRow();
+                if (!row)
+                    break;
+                addRow(row.getClear());
+            }
         }
+        if (!stopping)
+            prepareRHS();
     }
     void prepareRHS()
     {

thorlcr/activities/loop/thloopslave.cpp

@@ -286,7 +286,7 @@ public:
             }
         };
         ActPrintLog("maxIterations = %d", maxIterations);
-        dataLinkStart("LOOP", container.queryId());
+        dataLinkStart();
         nextRowFeeder.setown(new CNextRowFeeder(this, new CWrapper(*this)));
     }
     void doStop()
@@ -455,7 +455,7 @@ public:
         loopResults.setown(queryGraph().createThorGraphResults(0));
         helper->createParentExtract(extractBuilder);
         ActPrintLog("maxIterations = %d", maxIterations);
-        dataLinkStart("GRAPHLOOP", container.queryId());
+        dataLinkStart();
     }
     CATCH_NEXTROW()
     {
@@ -543,7 +543,7 @@ public:
         Owned<CGraphBase> graph = queryJob().getGraph(resultGraphId);
         Owned<IThorResult> result = graph->getResult(helper->querySequence(), queryGraph().isLocalChild());
         resultStream.setown(result->getRowStream());
-        dataLinkStart("LOCALRESULTREAD", container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {
@@ -622,7 +622,7 @@ public:
         IThorResult *result = graph->createResult(*this, helper->querySequence(), this, !queryGraph().isLocalChild());  // NB graph owns result
         resultWriter.setown(result->getWriter());
         startInput(inputs.item(0));
-        dataLinkStart("LOCALRESULTSPILL", container.queryId());
+        dataLinkStart();
     }
     CATCH_NEXTROW()
     {
@@ -820,7 +820,7 @@ public:
         selectedInput = container.whichBranch>=inputs.ordinality() ? NULL : inputs.item(container.whichBranch);
         if (selectedInput)
             startInput(selectedInput);
-        dataLinkStart(container.getKind(), container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {
@@ -886,7 +886,7 @@ public:
         started = false;
         eos = false;
         ok = false;
-        dataLinkStart("CHILDNORMALIZE", container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {
@@ -959,7 +959,7 @@ public:
     {
         ActivityTimer s(totalCycles, timeActivities, NULL);
         eos = false;
-        dataLinkStart("CHILDAGGREGATE", container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {
@@ -1016,7 +1016,7 @@ public:
         aggregated.clear();
         aggregated.setown(new CThorRowAggregator(*this, *helper, *helper));
         aggregated->start(queryRowAllocator());
-        dataLinkStart("CHILDGROUPAGGREGATE", container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {
@@ -1091,7 +1091,7 @@ public:
         lastInput.clear();
         nextOutput.clear();
         startInput(inputs.item(0));
-        dataLinkStart("CHILDTHROUGHNORMALIZE", container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {
@@ -1189,7 +1189,7 @@ public:
         }
         else
             abortSoon = true;
-        dataLinkStart("GRAPHLOOPRESULTREAD", container.queryId());
+        dataLinkStart();
     }
     virtual bool isGrouped() { return false; }
     CATCH_NEXTROW()

thorlcr/activities/merge/thmergeslave.cpp

@@ -338,7 +338,7 @@ public:
         writer.clear();
         out.setown(createPartitionMerger(sample));
 
-        dataLinkStart("MERGE", container.queryId());
+        dataLinkStart();
     }
 
     virtual void stop()
@@ -455,7 +455,7 @@ public:
         Owned<IRowLinkCounter> linkcounter = new CThorRowLinkCounter;
         out.setown(createRowStreamMerger(streams.ordinality(), streams.getArray(), helper->queryCompare(), helper->dedup(), linkcounter));
 
-        dataLinkStart("MERGE", container.queryId());
+        dataLinkStart();
     }
 
     virtual void stop()

thorlcr/activities/msort/thgroupsortslave.cpp

@@ -57,7 +57,7 @@ public:
     void start()
     {
         ActivityTimer s(totalCycles, timeActivities, NULL);
-        dataLinkStart(activityKindStr(queryContainer().getKind()), container.queryId());
+        dataLinkStart();
         input = inputs.item(0);
         unsigned spillPriority = container.queryGrouped() ? SPILL_PRIORITY_GROUPSORT : SPILL_PRIORITY_LARGESORT;
         iLoader.setown(createThorRowLoader(*this, queryRowInterfaces(input), iCompare, !unstable, rc_mixed, spillPriority));
@@ -137,7 +137,7 @@ public:
     void start()
     {
         ActivityTimer s(totalCycles, timeActivities, NULL);
-        dataLinkStart("SORTED", container.queryId());
+        dataLinkStart();
         input = inputs.item(0);
         startInput(input);
         IInputSteppingMeta *stepMeta = input->querySteppingMeta();

thorlcr/activities/msort/thmsortslave.cpp

@@ -103,7 +103,7 @@ public:
                 barrier->cancel();
                 throw;
             }
-            dataLinkStart("SORT", container.queryId());
+            dataLinkStart();
             
             Linked<IRowInterfaces> rowif = queryRowInterfaces(input);
             Owned<IRowInterfaces> auxrowif = createRowInterfaces(helper->querySortedRecordSize(),queryActivityId(),queryCodeContext());

thorlcr/activities/normalize/thnormalizeslave.cpp

@@ -58,7 +58,7 @@ public:
         anyThisGroup = false;
         input = inputs.item(0);
         startInput(input);
-        dataLinkStart("NORMALIZE", container.queryId()); 
+        dataLinkStart();
     }
     void stop()
     { 
@@ -141,7 +141,7 @@ public:
         input = inputs.item(0);
         startInput(input);
         anyThisGroup = false;
-        dataLinkStart("NORMALIZECHILD", container.queryId()); 
+        dataLinkStart();
         curChildRow = NULL;
     }
     void stop()
@@ -239,7 +239,7 @@ public:
         input = inputs.item(0);
         startInput(input);
         anyThisGroup = false;
-        dataLinkStart("NORMALIZECHILD", container.queryId()); 
+        dataLinkStart();
     }
     void stop()
     {

thorlcr/activities/nsplitter/thnsplitterslave.cpp

@@ -211,7 +211,7 @@ class NSplitterSlaveActivity : public CSlaveActivity
         {
             activity->ensureInputsConfigured();
             input->start();
-            dataLinkStart("SPLITTEROUTPUT", activity->queryContainer().queryId(), id);
+            dataLinkStart(id);
         }
         virtual bool isGrouped() { return activity->inputs.item(0)->isGrouped(); }
         virtual void getMetaInfo(ThorDataLinkMetaInfo &info)

thorlcr/activities/null/thnullslave.cpp

@@ -59,7 +59,7 @@ public:
     virtual void start()
     {
         ActivityTimer s(totalCycles, timeActivities, NULL);
-        dataLinkStart("NULL", container.queryId());
+        dataLinkStart();
     }
 
     virtual void stop()
@@ -106,7 +106,7 @@ public:
     {
         ActivityTimer s(totalCycles, timeActivities, NULL);
         startInput(inputs.item(0));
-        dataLinkStart("NULL", container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {

thorlcr/activities/nullaction/thnullactionslave.cpp

@@ -40,7 +40,7 @@ public:
     void start()
     {
         ActivityTimer s(totalCycles, timeActivities, NULL);
-        dataLinkStart("NULLACTION", container.queryId());
+        dataLinkStart();
     }
     void stop() { dataLinkStop(); }
 

thorlcr/activities/parse/thparseslave.cpp

@@ -71,7 +71,7 @@ public:
         ActivityTimer s(totalCycles, timeActivities, NULL);
         input = inputs.item(0);
         startInput(input);
-        dataLinkStart("ParseSlaveActivity", container.queryId());
+        dataLinkStart();
     }
     void stop()
     { 

thorlcr/activities/piperead/thprslave.cpp

@@ -253,7 +253,7 @@ public:
         eof = false;
         OwnedRoxieString pipeProgram(helper->getPipeProgram());
         openPipe(pipeProgram, "PIPEREAD");
-        dataLinkStart("PIPEREAD", container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {
@@ -381,7 +381,7 @@ public:
             openPipe(pipeProgram, "PIPETHROUGH");
         }
         startInput(inputs.item(0));
-        dataLinkStart("PIPETHROUGH", container.queryId());
+        dataLinkStart();
         pipeWriter = new PipeWriterThread(*this);
         pipeWriter->start();
     }

thorlcr/activities/project/thprojectslave.cpp

@@ -46,7 +46,7 @@ public:
         ActivityTimer s(totalCycles, timeActivities, NULL);
         input = inputs.item(0);
         startInput(input);
-        dataLinkStart("PROJECT", container.queryId());
+        dataLinkStart();
     }
     void stop()
     {
@@ -262,7 +262,7 @@ public:
         eof = !helper->canMatchAny();
         if (parallel)
             prefetcher.start();
-        dataLinkStart("PREFETCHPROJECT", container.queryId());
+        dataLinkStart();
     }
     void stop()
     {

thorlcr/activities/pull/thpullslave.cpp

@@ -47,7 +47,7 @@ public:
         ActivityTimer s(totalCycles, timeActivities, NULL);
         input.setown(createDataLinkSmartBuffer(this,inputs.item(0),PULL_SMART_BUFFER_SIZE,true,false,RCUNBOUND,NULL,false,&container.queryJob().queryIDiskUsage()));
         startInput(input);
-        dataLinkStart("PULL", container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {

thorlcr/activities/rollup/throllupslave.cpp

@@ -150,7 +150,6 @@ class CDedupRollupBaseActivity : public CSlaveActivity, implements IStopInput
     Linked<IRowInterfaces> rowif;
 
 protected:
-    StringAttr id;
     bool eogNext, eos;
     bool global;
     bool groupOp;
@@ -167,10 +166,6 @@ public:
         rollup = _rollup;
         global = _global;
         groupOp = _groupOp;
-        StringBuffer tmp;
-        tmp.append(global?"GLOBAL":groupOp?"GROUP":"LOCAL");
-        tmp.append(rollup?"ROLLUP":"DEDUP");
-        id.set(tmp.str());
     }
     virtual void stopInput()
     {
@@ -306,7 +301,7 @@ public:
     {
         ActivityTimer s(totalCycles, timeActivities, NULL);
         CDedupRollupBaseActivity::start();
-        dataLinkStart(id, container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {
@@ -493,7 +488,7 @@ public:
     {
         ActivityTimer s(totalCycles, timeActivities, NULL);
         CDedupRollupBaseActivity::start();
-        dataLinkStart(id, container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {
@@ -586,7 +581,7 @@ public:
         input = inputs.item(0);
         eoi = false;
         startInput(input);
-        dataLinkStart("ROLLUPGROUP", container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {

thorlcr/activities/sample/thsampleslave.cpp

@@ -50,7 +50,7 @@ public:
         numSamples = helper->getProportion();
         whichSample = helper->getSampleNumber();
         numToSkip = whichSample ? whichSample - 1 : 0;
-        dataLinkStart("SAMPLE", container.queryId());
+        dataLinkStart();
     }
 
 

thorlcr/activities/selectnth/thselectnthslave.cpp

@@ -103,7 +103,7 @@ public:
             sendN();
             throw;
         }
-        dataLinkStart("SELECTNTH", container.queryId());
+        dataLinkStart();
 
         seenNth = false;
         if (0==helper->getRowToSelect())

thorlcr/activities/selfjoin/thselfjoinslave.cpp

@@ -163,7 +163,7 @@ public:
         ActivityTimer s(totalCycles, timeActivities, NULL);
         input = inputs.item(0);
         startInput(input);
-        dataLinkStart("SELFJOIN", container.queryId());
+        dataLinkStart();
         bool hintparallelmatch = container.queryXGMML().getPropInt("hint[@name=\"parallel_match\"]/@value")!=0;
         bool hintunsortedoutput = container.queryXGMML().getPropInt("hint[@name=\"unsorted_output\"]/@value")!=0;
 

thorlcr/activities/soapcall/thsoapcallslave.cpp

@@ -68,7 +68,7 @@ public:
     {
         ActivityTimer s(totalCycles, timeActivities, NULL);
         eof = false;
-        dataLinkStart(container.getKind(), container.queryId());
+        dataLinkStart();
         if (wscHelper)
             wscHelper->start();
     }
@@ -148,7 +148,7 @@ public:
         eof = false;
         input = inputs.item(0);
         startInput(input);
-        dataLinkStart(container.getKind(), container.queryId());
+        dataLinkStart();
         wscHelper->start();
     }
     virtual void stop()

thorlcr/activities/spill/thspillslave.cpp

@@ -162,7 +162,7 @@ public:
         input = inputs.item(0);
         startInput(input);
         
-        dataLinkStart("SPILL", container.queryId());
+        dataLinkStart();
 
         open();
         hadrow = false;

thorlcr/activities/temptable/thtmptableslave.cpp

@@ -56,7 +56,7 @@ public:
     void start()
     {
         ActivityTimer s(totalCycles, timeActivities, NULL);
-        dataLinkStart("InlineTABLE", container.queryId());
+        dataLinkStart();
         __uint64 numRows = helper->numRows();
         // local when generated from a child query (the range is per node, don't split)
         bool isLocal = container.queryOwnerId() && container.queryOwner().isLocalOnly();

thorlcr/activities/thactivityutil.ipp

@@ -91,30 +91,13 @@ class CThorDataLink : implements IThorDataLink
 {
     CActivityBase *owner;
     rowcount_t count, icount;
-    char *activityName;
-    activity_id activityId;
     unsigned outputId;
     unsigned limit;
 
 protected:
-    inline void dataLinkStart()
+    inline void dataLinkStart(unsigned _outputId = 0)
     {
-        dataLinkStart(NULL, 0);
-    }
-
-    inline void dataLinkStart(ThorActivityKind kind, activity_id activityId, unsigned outputId = 0)
-    {
-        dataLinkStart(activityKindStr(kind), activityId, outputId);
-    }
-    inline void dataLinkStart(const char * _activityName, activity_id _activityId, unsigned _outputId = 0)
-    {
-        if(_activityName) 
-        {
-            StringBuffer x(_activityName);
-            activityName = x.toUpperCase().detach();
-            activityId = _activityId;
-            outputId = _outputId;
-        }
+        outputId = _outputId;
 #ifdef _TESTING
         ActPrintLog(owner, "ITDL starting for output %d", outputId);
 #endif
@@ -136,11 +119,6 @@ protected:
 #ifdef _TESTING
         ActPrintLog(owner, "ITDL output %d stopped, count was %"RCPF"d", outputId, getDataLinkCount());
 #endif
-        if(activityName) 
-        {
-            free(activityName);
-            activityName = NULL;
-        }
     }
 
     inline void dataLinkIncrement()
@@ -178,8 +156,6 @@ public:
     CThorDataLink(CActivityBase *_owner) : owner(_owner)
     {
         icount = count = 0;
-        activityName = NULL;
-        activityId = 0;
     }
 #ifdef _TESTING
     ~CThorDataLink()
@@ -223,6 +199,7 @@ interface ISmartBufferNotify
     virtual void onInputFinished(rowcount_t count) =0;
 };
 
+
 class CThorRowAggregator : public RowAggregator
 {
     CActivityBase &activity;

thorlcr/activities/topn/thtopnslave.cpp

@@ -258,7 +258,7 @@ public:
             eos = false;
         }
         eog = false;
-        dataLinkStart("TOPN", container.queryId());
+        dataLinkStart();
     }
     virtual bool isGrouped() { return grouped; }
     void stop()

thorlcr/activities/when/thwhenslave.cpp

@@ -93,7 +93,7 @@ public:
         ActivityTimer s(totalCycles, timeActivities, NULL);
         input.set(inputs.item(0));
         startInput(input);
-        dataLinkStart("WHEN", container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {

thorlcr/activities/wuidread/thwuidreadslave.cpp

@@ -54,7 +54,7 @@ public:
     void start()
     {
         ActivityTimer s(totalCycles, timeActivities, NULL);
-        dataLinkStart("WUIDREAD", container.queryId());
+        dataLinkStart();
 
         eogPending = false;
         if (container.queryLocal() || firstNode())

thorlcr/activities/xmlparse/thxmlparseslave.cpp

@@ -71,7 +71,7 @@ public:
         eogNext = false;
         input = inputs.item(0);
         startInput(input);
-        dataLinkStart("XMLPARSE", container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {

thorlcr/activities/xmlread/thxmlreadslave.cpp

@@ -214,7 +214,7 @@ public:
         ActivityTimer s(totalCycles, timeActivities, NULL);
         CDiskReadSlaveActivityBase::start();
         out = createSequentialPartHandler(partHandler, partDescs, false);
-        dataLinkStart("XMLREAD", container.queryId());
+        dataLinkStart();
     }
     virtual void stop()
     {