HPCC-14409 add a activity getRowAllocator queryRowManager

cosmetic change.

Add CActivityBase::getRowAllocator and a
CActivityBase::queryRowManager, for slightly more readibility in
activity code.

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

thorlcr/activities/aggregate/thaggregate.cpp

@@ -74,7 +74,7 @@ public:
             if (sz)
             {
                 IHThorThroughAggregateArg *helper = (IHThorThroughAggregateArg *)queryHelper();
-                Owned<IRowInterfaces> aggRowIf = createRowInterfaces(helper->queryAggregateRecordSize(), queryActivityId(), queryCodeContext());
+                Owned<IRowInterfaces> aggRowIf = createRowInterfaces(helper->queryAggregateRecordSize(), queryId(), queryCodeContext());
                 CThorStreamDeserializerSource mds(sz, msg.readDirect(sz));
                 RtlDynamicRowBuilder rowBuilder(aggRowIf->queryRowAllocator());
                 size32_t sz = aggRowIf->queryRowDeserializer()->deserialize(rowBuilder, mds);

thorlcr/activities/aggregate/thaggregateslave.cpp

@@ -267,7 +267,7 @@ public:
     {
         ActivityTimer s(totalCycles, timeActivities);
         doStart();
-        aggrowif.setown(createRowInterfaces(helper->queryAggregateRecordSize(),queryActivityId(),queryCodeContext()));
+        aggrowif.setown(createRowInterfaces(helper->queryAggregateRecordSize(),queryId(),queryCodeContext()));
         partResult.setAllocator(aggrowif->queryRowAllocator()).ensureRow();
         helper->clearAggregate(partResult);
         dataLinkStart();

thorlcr/activities/diskread/thdiskread.cpp

@@ -117,7 +117,7 @@ public:
     {
         IRecordSize *recordSize = helper->queryOutputMeta();
 
-        Owned<IRowInterfaces> rowIf = createRowInterfaces(helper->queryOutputMeta(), queryActivityId(), queryCodeContext());                
+        Owned<IRowInterfaces> rowIf = createRowInterfaces(helper->queryOutputMeta(), queryId(), queryCodeContext());                
         OwnedConstThorRow result = getAggregate(*this, container.queryJob().querySlaves(), *rowIf, *helper, mpTag);
         if (!result)
             return;

thorlcr/activities/fetch/thfetchslave.cpp

@@ -341,12 +341,12 @@ public:
             memset(encryptedKey, 0, encryptedKeyLen);
             free(encryptedKey);
         }
-        fetchDiskRowIf.setown(createRowInterfaces(fetchContext->queryDiskRecordSize(),queryActivityId(),queryCodeContext()));
+        fetchDiskRowIf.setown(createRowInterfaces(fetchContext->queryDiskRecordSize(),queryId(),queryCodeContext()));
         if (fetchBaseHelper->extractAllJoinFields())
         {
             IOutputMetaData *keyRowMeta = QUERYINTERFACE(fetchBaseHelper->queryExtractedSize(), IOutputMetaData);
             assertex(keyRowMeta);
-            keyRowAllocator.setown(queryJobChannel().getRowAllocator(keyRowMeta, queryActivityId()));
+            keyRowAllocator.setown(getRowAllocator(keyRowMeta));
         }
         appendOutputLinked(this);
     }
@@ -420,7 +420,7 @@ public:
                 keyIn = new CKeyFieldExtract(this, *in, *fetchBaseHelper, *fetchContext);
                 keyInMeta.set(QUERYINTERFACE(fetchBaseHelper->queryExtractedSize(), IOutputMetaData));
             }
-            keyInIf.setown(createRowInterfaces(keyInMeta,queryActivityId(),queryCodeContext()));
+            keyInIf.setown(createRowInterfaces(keyInMeta,queryId(),queryCodeContext()));
         }
         else
         {
@@ -449,11 +449,11 @@ public:
                 }
             };
             Owned<IOutputMetaData> fmeta = createFixedSizeMetaData(sizeof(offset_t)); // should be provided by Gavin?
-            keyInIf.setown(createRowInterfaces(fmeta,queryActivityId(),queryCodeContext()));
+            keyInIf.setown(createRowInterfaces(fmeta,queryId(),queryCodeContext()));
             keyIn = new CKeyFPosExtract(keyInIf, this, *in, *fetchBaseHelper, *fetchContext);
         }
 
-        Owned<IRowInterfaces> rowIf = createRowInterfaces(queryRowMetaData(), queryActivityId(), queryCodeContext());
+        Owned<IRowInterfaces> rowIf = createRowInterfaces(queryRowMetaData(), queryId(), queryCodeContext());
         fetchStream = createFetchStream(*this, keyInIf, rowIf, abortSoon, parts, offsetCount, offsetMapSz, offsetMapBytes.toByteArray(), this, mptag, eexp);
         fetchStreamOut = fetchStream->queryOutput();
         fetchStream->start(keyIn);

thorlcr/activities/hashdistrib/thhashdistribslave.cpp

@@ -94,7 +94,6 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
     IStopInput *istop;
     size32_t fixedEstSize;
     Owned<IRowWriter> pipewr;
-    roxiemem::IRowManager *rowManager;
     Owned<ISmartRowBuffer> piperd;
 
 protected:
@@ -1004,7 +1003,6 @@ public:
         pullBufferSize = DISTRIBUTE_PULL_BUFFER_SIZE;
         selfstopped = false;
         pull = false;
-        rowManager = activity->queryJobChannel().queryRowManager();
 
         StringBuffer compType;
         activity->getOpt(THOROPT_HDIST_COMP, compType);
@@ -2855,7 +2853,7 @@ public:
         iCompare = helper->queryCompare();
 
         // JCSMORE - really should ask / lookup what flags the allocator created for extractKey has...
-        allocFlags = queryJobChannel().queryThorAllocator()->queryFlags();
+        allocFlags = queryJobChannel().queryThorAllocator().queryFlags();
 
         // JCSMORE - it may not be worth extracting the key,
         // if there's an upstream activity that holds onto rows for periods of time (e.g. sort)
@@ -2867,16 +2865,16 @@ public:
             isVariable = km->isVariableSize();
             if (!isVariable && helper->queryOutputMeta()->isFixedSize())
             {
-                roxiemem::IRowManager *rM = queryJobChannel().queryRowManager();
-                memsize_t keySize = rM->getExpectedCapacity(km->getMinRecordSize(), allocFlags);
-                memsize_t rowSize = rM->getExpectedCapacity(helper->queryOutputMeta()->getMinRecordSize(), allocFlags);
+                roxiemem::IRowManager &rM = queryRowManager();
+                memsize_t keySize = rM.getExpectedCapacity(km->getMinRecordSize(), allocFlags);
+                memsize_t rowSize = rM.getExpectedCapacity(helper->queryOutputMeta()->getMinRecordSize(), allocFlags);
                 if (keySize >= rowSize)
                     extractKey = false;
             }
         }
         if (extractKey)
         {
-            _keyRowInterfaces.setown(createRowInterfaces(km, queryActivityId(), queryCodeContext()));
+            _keyRowInterfaces.setown(createRowInterfaces(km, queryId(), queryCodeContext()));
             keyRowInterfaces = _keyRowInterfaces;
             rowKeyCompare = helper->queryRowKeyCompare();
             iKeyHash = helper->queryKeyHash();
@@ -3023,7 +3021,7 @@ CHashTableRowTable::CHashTableRowTable(HashDedupSlaveActivityBase &_activity, IR
 void CHashTableRowTable::init(rowidx_t sz)
 {
     // reinitialize if need bigger or if requested size is much smaller than existing
-    rowidx_t newMaxRows = activity.queryJobChannel().queryRowManager()->getExpectedCapacity(sz * sizeof(rowidx_t *), activity.allocFlags) / sizeof(rowidx_t *);
+    rowidx_t newMaxRows = activity.queryRowManager().getExpectedCapacity(sz * sizeof(rowidx_t *), activity.allocFlags) / sizeof(rowidx_t *);
     if (newMaxRows <= maxRows && ((maxRows-newMaxRows) <= HASHDEDUP_HT_INC_SIZE))
         return;
     ReleaseThorRow(rows);
@@ -3087,7 +3085,7 @@ CBucket::CBucket(HashDedupSlaveActivityBase &_owner, IRowInterfaces *_rowIf, IRo
      */
     if (extractKey)
     {
-        _keyAllocator.setown(owner.queryJobChannel().getRowAllocator(keyIf->queryRowMetaData(), owner.queryActivityId(), owner.allocFlags));
+        _keyAllocator.setown(owner.getRowAllocator(keyIf->queryRowMetaData(), owner.allocFlags));
         keyAllocator = _keyAllocator;
     }
     else
@@ -3277,16 +3275,16 @@ CBucketHandler::CBucketHandler(HashDedupSlaveActivityBase &_owner, IRowInterface
 
 CBucketHandler::~CBucketHandler()
 {
-    owner.queryJobChannel().queryRowManager()->removeRowBuffer(this);
-    owner.queryJobChannel().queryRowManager()->removeRowBuffer(&postSpillFlush);
+    owner.queryRowManager().removeRowBuffer(this);
+    owner.queryRowManager().removeRowBuffer(&postSpillFlush);
     for (unsigned i=0; i<numBuckets; i++)
         ::Release(buckets[i]);
 }
 
 void CBucketHandler::flushBuckets()
 {
-    owner.queryJobChannel().queryRowManager()->removeRowBuffer(this);
-    owner.queryJobChannel().queryRowManager()->removeRowBuffer(&postSpillFlush);
+    owner.queryRowManager().removeRowBuffer(this);
+    owner.queryRowManager().removeRowBuffer(&postSpillFlush);
     for (unsigned i=0; i<numBuckets; i++)
     {
         CBucket &bucket = *buckets[i];
@@ -3335,11 +3333,11 @@ unsigned CBucketHandler::getBucketEstimate(rowcount_t totalRows) const
         // likely to be way off for variable
 
         // JCSMORE - will need to change based on whether upstream keeps packed or not.
-        roxiemem::IRowManager *rM = owner.queryJobChannel().queryRowManager();
+        roxiemem::IRowManager &rM = owner.queryRowManager();
 
         memsize_t availMem = roxiemem::getTotalMemoryLimit()-0x500000;
-        memsize_t initKeySize = rM->getExpectedCapacity(keyIf->queryRowMetaData()->getMinRecordSize(), owner.allocFlags);
-        memsize_t minBucketSpace = retBuckets * rM->getExpectedCapacity(HASHDEDUP_HT_BUCKET_SIZE * sizeof(void *), owner.allocFlags);
+        memsize_t initKeySize = rM.getExpectedCapacity(keyIf->queryRowMetaData()->getMinRecordSize(), owner.allocFlags);
+        memsize_t minBucketSpace = retBuckets * rM.getExpectedCapacity(HASHDEDUP_HT_BUCKET_SIZE * sizeof(void *), owner.allocFlags);
 
         rowcount_t _maxRowGuess = (availMem-minBucketSpace) / initKeySize; // without taking into account ht space / other overheads
         rowidx_t maxRowGuess;
@@ -3347,7 +3345,7 @@ unsigned CBucketHandler::getBucketEstimate(rowcount_t totalRows) const
             maxRowGuess = (rowidx_t)RCIDXMAX/sizeof(void *);
         else
             maxRowGuess = (rowidx_t)_maxRowGuess;
-        memsize_t bucketSpace = retBuckets * rM->getExpectedCapacity(((maxRowGuess+retBuckets-1)/retBuckets) * sizeof(void *), owner.allocFlags);
+        memsize_t bucketSpace = retBuckets * rM.getExpectedCapacity(((maxRowGuess+retBuckets-1)/retBuckets) * sizeof(void *), owner.allocFlags);
         // now rebase maxRowGuess
         _maxRowGuess = (availMem-bucketSpace) / initKeySize;
         if (_maxRowGuess >= RCIDXMAX/sizeof(void *))
@@ -3382,9 +3380,9 @@ void CBucketHandler::init(unsigned _numBuckets, IRowStream *keyStream)
         htRows.setOwner(buckets[i]);
     }
     ActPrintLog(&owner, "Max %d buckets, current depth = %d", numBuckets, depth+1);
-    owner.queryJobChannel().queryRowManager()->addRowBuffer(this);
+    owner.queryRowManager().addRowBuffer(this);
     // postSpillFlush not needed until after 1 spill event, but not safe to add within callback
-    owner.queryJobChannel().queryRowManager()->addRowBuffer(&postSpillFlush);
+    owner.queryRowManager().addRowBuffer(&postSpillFlush);
     if (keyStream)
     {
         loop
@@ -3772,7 +3770,7 @@ RowAggregator *mergeLocalAggs(Owned<IHashDistributor> &distributor, CActivityBas
             virtual void stop() { }
         };
         Owned<IOutputMetaData> nodeRowMeta = createOutputMetaDataWithChildRow(activity.queryRowAllocator(), sizeof(size32_t));
-        Owned<IRowInterfaces> nodeRowMetaRowIf = createRowInterfaces(nodeRowMeta, activity.queryActivityId(), activity.queryCodeContext());
+        Owned<IRowInterfaces> nodeRowMetaRowIf = createRowInterfaces(nodeRowMeta, activity.queryId(), activity.queryCodeContext());
         Owned<IRowStream> localAggregatedStream = new CRowAggregatedStream(activity, nodeRowMetaRowIf, localAggTable);
         class CNodeCompare : implements ICompare, implements IHash
         {

thorlcr/activities/indexread/thindexread.cpp

@@ -445,7 +445,7 @@ public:
     {
         if (container.queryLocalOrGrouped())
             return;
-        Owned<IRowInterfaces> rowIf = createRowInterfaces(helper->queryOutputMeta(), queryActivityId(), queryCodeContext());                
+        Owned<IRowInterfaces> rowIf = createRowInterfaces(helper->queryOutputMeta(), queryId(), queryCodeContext());                
         OwnedConstThorRow result = getAggregate(*this, container.queryJob().querySlaves(), *rowIf, *helper, mpTag);
         if (!result)
             return;

thorlcr/activities/indexwrite/thindexwriteslave.cpp

@@ -306,7 +306,7 @@ public:
 
         ThorDataLinkMetaInfo info;
         inputs.item(0)->getMetaInfo(info);
-        outRowAllocator.setown(queryJobChannel().getRowAllocator(helper->queryDiskRecordSize(), container.queryId()));
+        outRowAllocator.setown(getRowAllocator(helper->queryDiskRecordSize()));
         if (refactor)
         {
             assertex(isLocal);

thorlcr/activities/iterate/thgroupiterateslave.cpp

@@ -139,7 +139,7 @@ public:
     {
         appendOutputLinked(this);   // adding 'me' to outputs array
         helper = static_cast <IHThorProcessArg *> (queryHelper());
-        rightrowif.setown(createRowInterfaces(helper->queryRightRecordSize(),queryActivityId(),queryCodeContext()));
+        rightrowif.setown(createRowInterfaces(helper->queryRightRecordSize(),queryId(),queryCodeContext()));
         rightAllocator.set(rightrowif->queryRowAllocator());
     }
     void start()

thorlcr/activities/iterate/thiterateslave.cpp

@@ -208,7 +208,7 @@ public:
     virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
     {
         helper = static_cast <IHThorProcessArg *> (queryHelper());
-        rightRowAllocator.setown(queryJobChannel().getRowAllocator(helper->queryRightRecordSize(),queryActivityId()));
+        rightRowAllocator.setown(getRowAllocator(helper->queryRightRecordSize()));
         IterateSlaveActivityBase::init(data,slaveData);
     }
 

thorlcr/activities/join/thjoin.cpp

@@ -184,10 +184,10 @@ public:
                 }
                 if (helper->getJoinFlags()&JFslidingmatch) // JCSMORE shouldn't be necessary
                     primaryKeySerializer = NULL;
-                Owned<IRowInterfaces> primaryRowIf = createRowInterfaces(primaryInput->queryHelper()->queryOutputMeta(), queryActivityId(), queryCodeContext());
+                Owned<IRowInterfaces> primaryRowIf = createRowInterfaces(primaryInput->queryHelper()->queryOutputMeta(), queryId(), queryCodeContext());
                 Owned<IRowInterfaces> secondaryRowIf;
                 if (secondaryInput)
-                    secondaryRowIf.setown(createRowInterfaces(secondaryInput->queryHelper()->queryOutputMeta(), queryActivityId(), queryCodeContext()));
+                    secondaryRowIf.setown(createRowInterfaces(secondaryInput->queryHelper()->queryOutputMeta(), queryId(), queryCodeContext()));
 
                 bool betweenjoin = (helper->getJoinFlags()&JFslidingmatch)!=0;
                 if (container.getKind() == TAKselfjoin)
@@ -229,7 +229,7 @@ public:
                 }
                 else if (!nosortPrimary()||betweenjoin)
                 {
-                    Owned<IRowInterfaces> secondaryRowIf = createRowInterfaces(secondaryInput->queryHelper()->queryOutputMeta(), queryActivityId(), queryCodeContext());
+                    Owned<IRowInterfaces> secondaryRowIf = createRowInterfaces(secondaryInput->queryHelper()->queryOutputMeta(), queryId(), queryCodeContext());
 
                     imaster->SortSetup(primaryRowIf, primaryCompare, primaryKeySerializer, false, true, NULL, NULL);
                     ActPrintLog("JOIN waiting for barrier.1");

thorlcr/activities/join/thjoinslave.cpp

@@ -604,8 +604,8 @@ public:
     CMergeJoinSlaveBaseActivity(CGraphElementBase *container, CMergeJoinProcessor &_processor) : CThorNarySlaveActivity(container), CThorDataLink(this), CThorSteppable(this), processor(_processor)
     {
         helper = (IHThorNWayMergeJoinArg *)queryHelper();
-        inputAllocator.setown(queryJobChannel().getRowAllocator(helper->queryInputMeta(), queryActivityId()));
-        outputAllocator.setown(queryJobChannel().getRowAllocator(helper->queryOutputMeta(), queryActivityId()));
+        inputAllocator.setown(getRowAllocator(helper->queryInputMeta()));
+        outputAllocator.setown(getRowAllocator(helper->queryOutputMeta()));
     }
     void init(MemoryBuffer &data, MemoryBuffer &slaveData)
     {

thorlcr/activities/keyedjoin/thkeyedjoinslave.cpp

@@ -737,7 +737,7 @@ class CKeyedJoinSlave : public CSlaveActivity, public CThorDataLink, implements
                     unsigned endRequestsCount = owner.container.queryJob().querySlaves();
                     Owned<IBitSet> endRequests = createThreadSafeBitSet(); // NB: verification only
 
-                    Owned<IRowInterfaces> fetchDiskRowIf = createRowInterfaces(owner.helper->queryDiskRecordSize(),owner.queryActivityId(),owner.queryCodeContext());
+                    Owned<IRowInterfaces> fetchDiskRowIf = createRowInterfaces(owner.helper->queryDiskRecordSize(),owner.queryId(),owner.queryCodeContext());
                     while (!aborted)
                     {
                         CMessageBuffer replyMb;
@@ -1832,7 +1832,7 @@ public:
         node = queryJobChannel().queryMyRank()-1;
         onFailTransform = (0 != (joinFlags & JFonfail)) && (0 == (joinFlags & JFmatchAbortLimitSkips));
 
-        joinFieldsAllocator.setown(queryJobChannel().getRowAllocator(helper->queryJoinFieldsRecordSize(), queryActivityId()));
+        joinFieldsAllocator.setown(getRowAllocator(helper->queryJoinFieldsRecordSize()));
         if (onFailTransform || (joinFlags & JFleftouter))
         {
             RtlDynamicRowBuilder rr(joinFieldsAllocator);
@@ -1970,16 +1970,16 @@ public:
                 Owned<IOutputMetaData> fetchInputMeta;
                 if (0 != helper->queryFetchInputRecordSize()->getMinRecordSize())
                 {
-                    fetchInputAllocator.setown(queryJobChannel().getRowAllocator(helper->queryFetchInputRecordSize(), queryActivityId()));
+                    fetchInputAllocator.setown(getRowAllocator(helper->queryFetchInputRecordSize()));
                     fetchInputMeta.setown(createOutputMetaDataWithChildRow(fetchInputAllocator, FETCHKEY_HEADER_SIZE));
                 }
                 else
                     fetchInputMeta.setown(createFixedSizeMetaData(FETCHKEY_HEADER_SIZE));
-                fetchInputMetaRowIf.setown(createRowInterfaces(fetchInputMeta,queryActivityId(),queryCodeContext()));
+                fetchInputMetaRowIf.setown(createRowInterfaces(fetchInputMeta,queryId(),queryCodeContext()));
                 fetchInputMetaAllocator.set(fetchInputMetaRowIf->queryRowAllocator());
 
                 Owned<IOutputMetaData> fetchOutputMeta = createOutputMetaDataWithChildRow(joinFieldsAllocator, FETCHKEY_HEADER_SIZE);
-                fetchOutputRowIf.setown(createRowInterfaces(fetchOutputMeta,queryActivityId(),queryCodeContext()));
+                fetchOutputRowIf.setown(createRowInterfaces(fetchOutputMeta,queryId(),queryCodeContext()));
 
                 fetchHandler = new CKeyedFetchHandler(*this);
 
@@ -2004,15 +2004,15 @@ public:
         if (needsDiskRead)
         {
             Owned<IOutputMetaData> meta = createFixedSizeMetaData(KEYLOOKUP_HEADER_SIZE);
-            keyLookupAllocator.setown(queryJobChannel().getRowAllocator(meta, queryActivityId()));
+            keyLookupAllocator.setown(getRowAllocator(meta));
         }
         else
         {
             Owned<IOutputMetaData> meta = createOutputMetaDataWithChildRow(joinFieldsAllocator, KEYLOOKUP_HEADER_SIZE);
-            keyLookupAllocator.setown(queryJobChannel().getRowAllocator(meta, queryActivityId()));
+            keyLookupAllocator.setown(getRowAllocator(meta));
         }
 
-        indexInputAllocator.setown(queryJobChannel().getRowAllocator(helper->queryIndexReadInputRecordSize(), queryActivityId()));
+        indexInputAllocator.setown(getRowAllocator(helper->queryIndexReadInputRecordSize()));
 
         ////////////////////
 

thorlcr/activities/lookupjoin/thlookupjoinslave.cpp

@@ -582,7 +582,7 @@ public:
         else
         {
             size32_t bitSetMemSz = getBitSetMemoryRequirement(rowCount);
-            void *pBitSetMem = activity.queryJobChannel().queryRowManager()->allocate(bitSetMemSz, activity.queryContainer().queryId(), SPILL_PRIORITY_LOW);
+            void *pBitSetMem = activity.queryRowManager().allocate(bitSetMemSz, activity.queryContainer().queryId(), SPILL_PRIORITY_LOW);
             if (!pBitSetMem)
                 return false;
 
@@ -1391,6 +1391,7 @@ protected:
     using PARENT::helper;
     using PARENT::clearHT;
     using PARENT::rhs;
+    using PARENT::queryRowManager;
 
     IHash *leftHash, *rightHash;
     ICompare *compareRight, *compareLeftRight;
@@ -1645,7 +1646,7 @@ protected:
                  * Need to remove spill callback and broadcast one last message to know.
                  */
 
-                queryJobChannel().queryRowManager()->removeRowBuffer(this);
+                queryRowManager().removeRowBuffer(this);
 
                 ActPrintLog("Broadcasting final split status");
                 broadcaster.reset();
@@ -2053,7 +2054,7 @@ public:
                 overflowWriteCount = 0;
                 overflowWriteFile.clear();
                 overflowWriteStream.clear();
-                queryJobChannel().queryRowManager()->addRowBuffer(this);
+                queryRowManager().addRowBuffer(this);
             }
         }
         else
@@ -2221,7 +2222,7 @@ public:
         memsize_t sz = (memsize_t)_sz;
         if (sz != _sz) // treat as OOM exception for handling purposes.
             throw MakeStringException(ROXIEMM_MEMORY_LIMIT_EXCEEDED, "Unsigned overflow, trying to allocate hash table of size: %" I64F "d ", _sz);
-        void *ht = activity->queryJobChannel().queryRowManager()->allocate(sz, activity->queryContainer().queryId(), SPILL_PRIORITY_LOW);
+        void *ht = activity->queryRowManager().allocate(sz, activity->queryContainer().queryId(), SPILL_PRIORITY_LOW);
         memset(ht, 0, sz);
         htMemory.setown(ht);
         htSize = size;

thorlcr/activities/loop/thloop.cpp

@@ -338,7 +338,7 @@ public:
         assertex(container.queryResultsGraph());
         Owned<CGraphBase> graph = queryJobChannel().getGraph(container.queryResultsGraph()->queryGraphId());
         IHThorLocalResultWriteArg *helper = (IHThorLocalResultWriteArg *)queryHelper();
-        inputRowIf.setown(createRowInterfaces(container.queryInput(0)->queryHelper()->queryOutputMeta(),queryActivityId(),queryCodeContext()));
+        inputRowIf.setown(createRowInterfaces(container.queryInput(0)->queryHelper()->queryOutputMeta(),queryId(),queryCodeContext()));
         createResult();
     }
 };

thorlcr/activities/merge/thmerge.cpp

@@ -58,7 +58,7 @@ public:
         ActPrintLog("GlobalMergeActivityMaster::process");
         CMasterActivity::process();     
         IHThorMergeArg *helper = (IHThorMergeArg *)queryHelper();   
-        Owned<IRowInterfaces> rowif = createRowInterfaces(helper->queryOutputMeta(),queryActivityId(),queryCodeContext());
+        Owned<IRowInterfaces> rowif = createRowInterfaces(helper->queryOutputMeta(),queryId(),queryCodeContext());
         CThorKeyArray sample(*this, rowif,helper->querySerialize(),helper->queryCompare(),helper->queryCompareKey(),helper->queryCompareRowKey());
 
         unsigned n = container.queryJob().querySlaves();

thorlcr/activities/msort/thmsort.cpp

@@ -155,8 +155,8 @@ protected:
                 skewThreshold = container.queryJob().getWorkUnitValueInt("defaultSkewThreshold", 0);
         }
 
-        Owned<IRowInterfaces> rowif = createRowInterfaces(container.queryInput(0)->queryHelper()->queryOutputMeta(),queryActivityId(),queryCodeContext());
-        Owned<IRowInterfaces> auxrowif = createRowInterfaces(helper->querySortedRecordSize(),queryActivityId(),queryCodeContext());
+        Owned<IRowInterfaces> rowif = createRowInterfaces(container.queryInput(0)->queryHelper()->queryOutputMeta(),queryId(),queryCodeContext());
+        Owned<IRowInterfaces> auxrowif = createRowInterfaces(helper->querySortedRecordSize(),queryId(),queryCodeContext());
         try
         {
             imaster->SortSetup(rowif,helper->queryCompare(),helper->querySerialize(),cosortfilenames.length()!=0,true,cosortfilenames.str(),auxrowif);

thorlcr/activities/msort/thmsortslave.cpp

@@ -106,7 +106,7 @@ public:
             dataLinkStart();
             
             Linked<IRowInterfaces> rowif = queryRowInterfaces(input);
-            Owned<IRowInterfaces> auxrowif = createRowInterfaces(helper->querySortedRecordSize(),queryActivityId(),queryCodeContext());
+            Owned<IRowInterfaces> auxrowif = createRowInterfaces(helper->querySortedRecordSize(),queryId(),queryCodeContext());
             sorter->Gather(
                 rowif,
                 input,

thorlcr/activities/msort/thsortu.cpp

@@ -1683,7 +1683,7 @@ class CMultiCoreJoinHelper: public CMultiCoreJoinHelperBase
             PROGLOG("CMultiCoreJoinHelper::cWorker started");
 
             Owned<IRowInterfaces> rowIf = parent->activity.getRowInterfaces();
-            Owned<IEngineRowAllocator> allocator = parent->activity.queryJobChannel().getRowAllocator(rowIf->queryRowMetaData(), parent->activity.queryActivityId(), (roxiemem::RoxieHeapFlags)(roxiemem::RHFpacked|roxiemem::RHFunique));
+            Owned<IEngineRowAllocator> allocator = parent->activity.getRowAllocator(rowIf->queryRowMetaData(), (roxiemem::RoxieHeapFlags)(roxiemem::RHFpacked|roxiemem::RHFunique));
 
             IRowWriter *rowWriter = rowStream->queryWriter();
             loop
@@ -1882,7 +1882,7 @@ class CMultiCoreUnorderedJoinHelper: public CMultiCoreJoinHelperBase
         int run()
         {
             Owned<IRowInterfaces> rowIf = parent->activity.getRowInterfaces();
-            Owned<IEngineRowAllocator> allocator = parent->activity.queryJobChannel().getRowAllocator(rowIf->queryRowMetaData(), parent->activity.queryActivityId(), (roxiemem::RoxieHeapFlags)(roxiemem::RHFpacked|roxiemem::RHFunique));
+            Owned<IEngineRowAllocator> allocator = parent->activity.getRowAllocator(rowIf->queryRowMetaData(), (roxiemem::RoxieHeapFlags)(roxiemem::RHFpacked|roxiemem::RHFunique));
 
             Owned<IRowWriter> rowWriter = parent->multiWriter->getWriter();
             PROGLOG("CMulticoreUnorderedJoinHelper::cWorker started");

thorlcr/activities/piperead/thprslave.cpp

@@ -238,7 +238,7 @@ public:
         IRowInterfaces *_inrowif;
         if (needTransform)
         {
-            inrowif.setown(createRowInterfaces(helper->queryDiskRecordSize(),queryActivityId(),queryCodeContext()));
+            inrowif.setown(createRowInterfaces(helper->queryDiskRecordSize(),queryId(),queryCodeContext()));
             _inrowif = inrowif;
         }
         else

thorlcr/activities/result/thresult.cpp

@@ -65,7 +65,7 @@ public:
                 if (results)
                     throw MakeThorException(TE_UnexpectedMultipleSlaveResults, "Received greater than one result from slaves");
                 IHThorRemoteResultArg *helper = (IHThorRemoteResultArg *)queryHelper();
-                Owned<IRowInterfaces> resultRowIf = createRowInterfaces(helper->queryOutputMeta(), queryActivityId(), queryCodeContext());
+                Owned<IRowInterfaces> resultRowIf = createRowInterfaces(helper->queryOutputMeta(), queryId(), queryCodeContext());
                 CThorStreamDeserializerSource mds(sz, mb.readDirect(sz));
                 RtlDynamicRowBuilder rowBuilder(resultRowIf->queryRowAllocator());
                 size32_t sz = resultRowIf->queryRowDeserializer()->deserialize(rowBuilder, mds);

thorlcr/activities/thdiskbaseslave.cpp

@@ -273,7 +273,7 @@ void CDiskReadSlaveActivityBase::kill()
 IRowInterfaces * CDiskReadSlaveActivityBase::queryDiskRowInterfaces()
 {
     if (!diskRowIf) 
-        diskRowIf.setown(createRowInterfaces(helper->queryDiskRecordSize(),queryActivityId(),queryCodeContext()));
+        diskRowIf.setown(createRowInterfaces(helper->queryDiskRecordSize(),queryId(),queryCodeContext()));
     return diskRowIf;
 }
 

thorlcr/activities/wuidwrite/thwuidwrite.cpp

@@ -303,7 +303,7 @@ public:
     {
         assertex(complete);
         ActPrintLog("dictionary result");
-        Owned<IRowInterfaces> rowIf = createRowInterfaces(container.queryInput(0)->queryHelper()->queryOutputMeta(),queryActivityId(),queryCodeContext());
+        Owned<IRowInterfaces> rowIf = createRowInterfaces(container.queryInput(0)->queryHelper()->queryOutputMeta(),queryId(),queryCodeContext());
         IOutputRowDeserializer *deserializer = rowIf->queryRowDeserializer();
         CMessageBuffer mb;
         Owned<ISerialStream> stream = createMemoryBufferSerialStream(resultData);

thorlcr/graph/thgraph.cpp

@@ -2687,7 +2687,7 @@ CJobChannel::CJobChannel(CJobBase &_job, IMPServer *_mpServer, unsigned _channel
 
 CJobChannel::~CJobChannel()
 {
-    thorAllocator->queryRowManager()->reportMemoryUsage(false);
+    queryRowManager().reportMemoryUsage(false);
     PROGLOG("CJobBase resetting memory manager");
     thorAllocator.clear();
     wait();
@@ -2724,12 +2724,12 @@ mptag_t CJobChannel::deserializeMPTag(MemoryBuffer &mb)
     return tag;
 }
 
-IEngineRowAllocator *CJobChannel::getRowAllocator(IOutputMetaData * meta, unsigned activityId, roxiemem::RoxieHeapFlags flags) const
+IEngineRowAllocator *CJobChannel::getRowAllocator(IOutputMetaData * meta, activity_id activityId, roxiemem::RoxieHeapFlags flags) const
 {
     return thorAllocator->getRowAllocator(meta, activityId, flags);
 }
 
-roxiemem::IRowManager *CJobChannel::queryRowManager() const
+roxiemem::IRowManager &CJobChannel::queryRowManager() const
 {
     return thorAllocator->queryRowManager();
 }
@@ -3048,7 +3048,7 @@ IEngineRowAllocator * CActivityBase::queryRowAllocator()
 {
     if (CABallocatorlock.lock()) {
         if (!rowAllocator)
-            rowAllocator.setown(queryJobChannel().getRowAllocator(queryRowMetaData(),queryActivityId()));
+            rowAllocator.setown(getRowAllocator(queryRowMetaData()));
         CABallocatorlock.unlock();
     }
     return rowAllocator;
@@ -3058,7 +3058,7 @@ IOutputRowSerializer * CActivityBase::queryRowSerializer()
 {
     if (CABserializerlock.lock()) {
         if (!rowSerializer)
-            rowSerializer.setown(queryRowMetaData()->createDiskSerializer(queryCodeContext(),queryActivityId()));
+            rowSerializer.setown(queryRowMetaData()->createDiskSerializer(queryCodeContext(),queryId()));
         CABserializerlock.unlock();
     }
     return rowSerializer;
@@ -3068,7 +3068,7 @@ IOutputRowDeserializer * CActivityBase::queryRowDeserializer()
 {
     if (CABdeserializerlock.lock()) {
         if (!rowDeserializer)
-            rowDeserializer.setown(queryRowMetaData()->createDiskDeserializer(queryCodeContext(),queryActivityId()));
+            rowDeserializer.setown(queryRowMetaData()->createDiskDeserializer(queryCodeContext(),queryId()));
         CABdeserializerlock.unlock();
     }
     return rowDeserializer;
@@ -3080,6 +3080,11 @@ IRowInterfaces *CActivityBase::getRowInterfaces()
     return createRowInterfaces(queryRowMetaData(), container.queryId(), queryCodeContext());
 }
 
+IEngineRowAllocator *CActivityBase::getRowAllocator(IOutputMetaData * meta, roxiemem::RoxieHeapFlags flags) const
+{
+    return queryJobChannel().getRowAllocator(meta, queryId(), flags);
+}
+
 bool CActivityBase::receiveMsg(CMessageBuffer &mb, const rank_t rank, const mptag_t mpTag, rank_t *sender, unsigned timeout)
 {
     BooleanOnOff onOff(receiving);

thorlcr/graph/thgraph.hpp

@@ -810,7 +810,6 @@ public:
     void addDependencies(IPropertyTree *xgmml, bool failIfMissing=true);
     void addSubGraph(IPropertyTree &xgmml);
 
-    roxiemem::IRowManager *queryRowManager() const;
     bool queryUseCheckpoints() const;
     bool queryPausing() const { return pausing; }
     bool queryResumed() const { return resumed; }
@@ -945,13 +944,13 @@ public:
 
     ICodeContext &queryCodeContext() const;
     IThorResult *getOwnedResult(graph_id gid, activity_id ownerId, unsigned resultId);
-    IThorAllocator *queryThorAllocator() const { return thorAllocator; }
+    IThorAllocator &queryThorAllocator() const { return *thorAllocator; }
     ICommunicator &queryJobComm() const { return *jobComm; }
     IMPServer &queryMPServer() const { return *mpServer; }
     const rank_t &queryMyRank() const { return myrank; }
     mptag_t deserializeMPTag(MemoryBuffer &mb);
-    IEngineRowAllocator *getRowAllocator(IOutputMetaData * meta, unsigned activityId, roxiemem::RoxieHeapFlags flags=roxiemem::RHFnone) const;
-    roxiemem::IRowManager *queryRowManager() const;
+    IEngineRowAllocator *getRowAllocator(IOutputMetaData * meta, activity_id activityId, roxiemem::RoxieHeapFlags flags=roxiemem::RHFnone) const;
+    roxiemem::IRowManager &queryRowManager() const;
 
     virtual void abort(IException *e);
     virtual IBarrier *createBarrier(mptag_t tag) { UNIMPLEMENTED; return NULL; }
@@ -989,14 +988,16 @@ public:
     IMPLEMENT_IINTERFACE;
     CActivityBase(CGraphElementBase *container);
     ~CActivityBase();
+    inline activity_id queryId() const { return container.queryId(); }
     CGraphElementBase &queryContainer() const { return container; }
     CJobBase &queryJob() const { return container.queryJob(); }
     CJobChannel &queryJobChannel() const { return container.queryJobChannel(); }
     inline IMPServer &queryMPServer() const { return queryJobChannel().queryMPServer(); }
+    inline roxiemem::IRowManager &queryRowManager() const { return queryJobChannel().queryRowManager(); }
     CGraphBase &queryGraph() const { return container.queryOwner(); }
     CActivityBase &queryChannelActivity(unsigned channel) const
     {
-        return queryJob().queryChannelActivity(channel, queryGraph().queryGraphId(), container.queryId());
+        return queryJob().queryChannelActivity(channel, queryGraph().queryGraphId(), queryId());
     }
     inline const mptag_t queryMpTag() const { return mpTag; }
     inline bool queryAbortSoon() const { return abortSoon; }
@@ -1010,6 +1011,7 @@ public:
     bool lastNode() { return container.queryJob().querySlaves() == container.queryJobChannel().queryMyRank(); }
     unsigned queryMaxCores() const { return maxCores; }
     IRowInterfaces *getRowInterfaces();
+    IEngineRowAllocator *getRowAllocator(IOutputMetaData * meta, roxiemem::RoxieHeapFlags flags=roxiemem::RHFnone) const;
 
     bool appendRowXml(StringBuffer & target, IOutputMetaData & meta, const void * row) const;
     void logRow(const char * prefix, IOutputMetaData & meta, const void * row);
@@ -1040,7 +1042,7 @@ public:
     virtual IOutputRowSerializer * queryRowSerializer(); 
     virtual IOutputRowDeserializer * queryRowDeserializer(); 
     virtual IOutputMetaData *queryRowMetaData() { return baseHelper->queryOutputMeta(); }
-    virtual unsigned queryActivityId() { return (unsigned)container.queryId(); }
+    virtual unsigned queryActivityId() { return (unsigned)queryId(); }
     virtual ICodeContext *queryCodeContext() { return container.queryCodeContext(); }
 
     StringBuffer &getOpt(const char *prop, StringBuffer &out) const;

thorlcr/slave/slave.cpp

@@ -778,7 +778,7 @@ activityslaves_decl IEngineRowAllocator * queryRowAllocator(IThorDataLink *link)
 activityslaves_decl IOutputRowSerializer * queryRowSerializer(IThorDataLink *link) { CActivityBase *base = link?link->queryFromActivity():NULL; return base?base->queryRowSerializer():NULL; }
 activityslaves_decl IOutputRowDeserializer * queryRowDeserializer(IThorDataLink *link) { CActivityBase *base = link?link->queryFromActivity():NULL; return base?base->queryRowDeserializer():NULL; }
 activityslaves_decl IOutputMetaData *queryRowMetaData(IThorDataLink *link) { CActivityBase *base = link?link->queryFromActivity():NULL; return base?base->queryRowMetaData():NULL; }
-activityslaves_decl unsigned queryActivityId(IThorDataLink *link) { CActivityBase *base = link?link->queryFromActivity():NULL; return base?base->queryActivityId():0; }
+activityslaves_decl unsigned queryActivityId(IThorDataLink *link) { CActivityBase *base = link?link->queryFromActivity():NULL; return base?base->queryId():0; }
 activityslaves_decl ICodeContext *queryCodeContext(IThorDataLink *link) { CActivityBase *base = link?link->queryFromActivity():NULL; return base?base->queryCodeContext():NULL; }
 
 

thorlcr/thorcodectx/thcodectx.cpp

@@ -115,12 +115,12 @@ IEngineRowAllocator * CThorCodeContextBase::getRowAllocator(IOutputMetaData * me
 
 const char * CThorCodeContextBase::cloneVString(const char * str) const
 {
-    return jobChannel.queryRowManager()->cloneVString(str);
+    return jobChannel.queryRowManager().cloneVString(str);
 }
 
 const char * CThorCodeContextBase::cloneVString(size32_t len, const char * str) const
 {
-    return jobChannel.queryRowManager()->cloneVString(len, str);
+    return jobChannel.queryRowManager().cloneVString(len, str);
 }
 
 IEclGraphResults *CThorCodeContextBase::resolveLocalQuery(__int64 gid)

thorlcr/thorutil/thbuf.cpp

@@ -1742,8 +1742,7 @@ public:
         if (readGranularity > limit)
             readGranularity = limit; // readGranularity must be <= limit;
         numWriters = 0;
-        roxiemem::IRowManager *rowManager = activity.queryJobChannel().queryRowManager();
-        readRows = static_cast<const void * *>(rowManager->allocate(readGranularity * sizeof(void*), activity.queryContainer().queryId()));
+        readRows = static_cast<const void * *>(activity.queryRowManager().allocate(readGranularity * sizeof(void*), activity.queryContainer().queryId()));
         eos = eow = readerBlocked = false;
         rowPos = rowsToRead = 0;
         writersComplete = writersBlocked = 0;

thorlcr/thorutil/thmem.cpp

@@ -177,7 +177,7 @@ protected:
             return false;
 
         StringBuffer tempName;
-        VStringBuffer tempPrefix("streamspill_%d", activity.queryActivityId());
+        VStringBuffer tempPrefix("streamspill_%d", activity.queryId());
         GetTempName(tempName, tempPrefix.str(), true);
         spillFile.setown(createIFile(tempName.str()));
 
@@ -191,7 +191,7 @@ protected:
         if (!mmRegistered)
         {
             mmRegistered = true;
-            activity.queryJobChannel().queryRowManager()->addRowBuffer(this);
+            activity.queryRowManager().addRowBuffer(this);
         }
     }
     inline void clearSpillingCallback()
@@ -199,7 +199,7 @@ protected:
         if (mmRegistered)
         {
             mmRegistered = false;
-            activity.queryJobChannel().queryRowManager()->removeRowBuffer(this);
+            activity.queryRowManager().removeRowBuffer(this);
         }
     }
 public:
@@ -359,8 +359,7 @@ public:
         granularity = 500; // JCSMORE - rows
 
         // a small amount of rows to read from swappable rows
-        roxiemem::IRowManager *rowManager = activity.queryJobChannel().queryRowManager();
-        readRows = static_cast<const void * *>(rowManager->allocate(granularity * sizeof(void*), activity.queryContainer().queryId(), inRows.queryDefaultMaxSpillCost()));
+        readRows = static_cast<const void * *>(activity.queryRowManager().allocate(granularity * sizeof(void*), activity.queryContainer().queryId(), inRows.queryDefaultMaxSpillCost()));
         addSpillingCallback();
     }
     ~CSpillableStream()
@@ -614,7 +613,7 @@ CThorExpandingRowArray::CThorExpandingRowArray(CActivityBase &_activity, IRowInt
     rows = NULL;
     maxRows = 0;
     numRows = 0;
-    rowManager = activity.queryJobChannel().queryRowManager();
+    rowManager = &activity.queryRowManager();
     throwOnOom = false;
     setup(_rowIf, _allowNulls, _stableSort, _throwOnOom);
     setDefaultMaxSpillCost(roxiemem::SpillAllCost);
@@ -1044,25 +1043,25 @@ offset_t CThorExpandingRowArray::serializedSize()
 
 memsize_t CThorExpandingRowArray::getMemUsage()
 {
-    roxiemem::IRowManager *rM = activity.queryJobChannel().queryRowManager();
+    roxiemem::IRowManager &rM = activity.queryRowManager();
     IOutputMetaData *meta = rowIf->queryRowMetaData();
     IOutputMetaData *diskMeta = meta->querySerializedDiskMeta(); // GH->JCS - really I want a internalMeta here.
     rowidx_t c = ordinality();
     memsize_t total = 0;
     if (diskMeta->isFixedSize())
-        total = c * rM->getExpectedFootprint(diskMeta->getFixedSize(), 0);
+        total = c * rM.getExpectedFootprint(diskMeta->getFixedSize(), 0);
     else
     {
         CSizingSerializer ssz;
         for (rowidx_t i=0; i<c; i++)
         {
             serializer->serialize(ssz, (const byte *)rows[i]);
-            total += rM->getExpectedFootprint(ssz.size(), 0);
+            total += rM.getExpectedFootprint(ssz.size(), 0);
             ssz.reset();
         }
     }
     // NB: worst case, when expanding (see resize method)
-    memsize_t sz = rM->getExpectedFootprint(maxRows * sizeof(void *), 0);
+    memsize_t sz = rM.getExpectedFootprint(maxRows * sizeof(void *), 0);
     memsize_t szE = sz / 100 * 125; // don't care if sz v. small
     if (stableSort_none == stableSort)
         total += sz + szE;
@@ -1503,7 +1502,7 @@ protected:
             ActPrintLog(&activity, "Sort took: %f", ((float)timer.elapsedMs())/1000);
             tempPrefix.append("srt");
         }
-        tempPrefix.appendf("spill_%d", activity.queryActivityId());
+        tempPrefix.appendf("spill_%d", activity.queryId());
         GetTempName(tempName, tempPrefix.str(), true);
         Owned<IFile> iFile = createIFile(tempName.str());
         VStringBuffer spillPrefixStr("RowCollector(%d)", spillPriority);
@@ -1663,7 +1662,7 @@ protected:
     {
         if (mmRegistered)
         {
-            activity.queryJobChannel().queryRowManager()->removeRowBuffer(this);
+            activity.queryRowManager().removeRowBuffer(this);
             mmRegistered = false;
         }
     }
@@ -1671,7 +1670,7 @@ protected:
     {
         if (!mmRegistered && spillingEnabled())
         {
-            activity.queryJobChannel().queryRowManager()->addRowBuffer(this);
+            activity.queryRowManager().addRowBuffer(this);
             mmRegistered = true;
         }
     }
@@ -1746,7 +1745,7 @@ public:
         if (mmRegistered && !spillingEnabled())
         {
             mmRegistered = false;
-            activity.queryJobChannel().queryRowManager()->removeRowBuffer(this);
+            activity.queryRowManager().removeRowBuffer(this);
         }
         spillableRows.setup(rowIf, false, stableSort);
     }
@@ -2126,17 +2125,17 @@ public:
         return createRoxieRowAllocator(cache, *rowManager, meta, activityId, id, flags);
     }
 // IThorAllocator
-    virtual IEngineRowAllocator *getRowAllocator(IOutputMetaData * meta, unsigned activityId, roxiemem::RoxieHeapFlags flags) const
+    virtual IEngineRowAllocator *getRowAllocator(IOutputMetaData * meta, activity_id activityId, roxiemem::RoxieHeapFlags flags) const
     {
         return allocatorMetaCache->ensure(meta, activityId, flags);
     }
-    virtual IEngineRowAllocator *getRowAllocator(IOutputMetaData * meta, unsigned activityId) const
+    virtual IEngineRowAllocator *getRowAllocator(IOutputMetaData * meta, activity_id activityId) const
     {
         return allocatorMetaCache->ensure(meta, activityId, defaultFlags);
     }
-    virtual roxiemem::IRowManager *queryRowManager() const
+    virtual roxiemem::IRowManager &queryRowManager() const
     {
-        return rowManager;
+        return *rowManager;
     }
     virtual roxiemem::RoxieHeapFlags queryFlags() const { return defaultFlags; }
     virtual bool queryCrc() const { return false; }

thorlcr/thorutil/thmem.hpp

@@ -150,9 +150,9 @@ private:
 
 interface IThorAllocator : extends IInterface
 {
-    virtual IEngineRowAllocator *getRowAllocator(IOutputMetaData * meta, unsigned activityId, roxiemem::RoxieHeapFlags flags) const = 0;
-    virtual IEngineRowAllocator *getRowAllocator(IOutputMetaData * meta, unsigned activityId) const = 0;
-    virtual roxiemem::IRowManager *queryRowManager() const = 0;
+    virtual IEngineRowAllocator *getRowAllocator(IOutputMetaData * meta, activity_id activityId, roxiemem::RoxieHeapFlags flags) const = 0;
+    virtual IEngineRowAllocator *getRowAllocator(IOutputMetaData * meta, activity_id activityId) const = 0;
+    virtual roxiemem::IRowManager &queryRowManager() const = 0;
     virtual roxiemem::RoxieHeapFlags queryFlags() const = 0;
     virtual bool queryCrc() const = 0;
 };