virtual void init()
 {
     CMasterActivity::init();
     OwnedRoxieString fname(helper->getFileName());
     Owned<IDistributedFile> fetchFile = queryThorFileManager().lookup(container.queryJob(), fname, false, 0 != (helper->getFetchFlags() & FFdatafileoptional), true);
     if (fetchFile)
     {
         if (isFileKey(fetchFile))
             throw MakeActivityException(this, 0, "Attempting to read index as a flat file: %s", fname.get());
         Owned<IFileDescriptor> fileDesc = getConfiguredFileDescriptor(*fetchFile);
         void *ekey;
         size32_t ekeylen;
         helper->getFileEncryptKey(ekeylen,ekey);
         bool encrypted = fileDesc->queryProperties().getPropBool("@encrypted");
         if (0 != ekeylen)
         {
             memset(ekey,0,ekeylen);
             free(ekey);
             if (!encrypted)
             {
                 Owned<IException> e = MakeActivityWarning(&container, TE_EncryptionMismatch, "Ignoring encryption key provided as file '%s' was not published as encrypted", fetchFile->queryLogicalName());
                 queryJobChannel().fireException(e);
             }
         }
         else if (encrypted)
             throw MakeActivityException(this, 0, "File '%s' was published as encrypted but no encryption key provided", fetchFile->queryLogicalName());
         mapping.setown(getFileSlaveMaps(fetchFile->queryLogicalName(), *fileDesc, container.queryJob().queryUserDescriptor(), container.queryJob().querySlaveGroup(), container.queryLocalOrGrouped(), false, NULL, fetchFile->querySuperFile()));
         mapping->serializeFileOffsetMap(offsetMapMb);
         addReadFile(fetchFile);
     }
 }

    void init()
    {
        helper = (IHThorKeyDiffArg *)queryHelper();
        OwnedRoxieString origName(helper->getOriginalName());
        OwnedRoxieString updatedName(helper->getUpdatedName());
        originalIndexFile.setown(queryThorFileManager().lookup(container.queryJob(), origName));
        newIndexFile.setown(queryThorFileManager().lookup(container.queryJob(), updatedName));
        if (originalIndexFile->numParts() != newIndexFile->numParts())
            throw MakeActivityException(this, TE_KeyDiffIndexSizeMismatch, "Index %s and %s differ in width", origName.get(), updatedName.get());
        if (originalIndexFile->querySuperFile() || newIndexFile->querySuperFile())
            throw MakeActivityException(this, 0, "Diffing super files not supported");  

        width = originalIndexFile->numParts();

        originalDesc.setown(originalIndexFile->getFileDescriptor());
        newIndexDesc.setown(newIndexFile->getFileDescriptor());
        Owned<IPartDescriptor> tlkDesc = originalDesc->getPart(originalDesc->numParts()-1);
        const char *kind = tlkDesc->queryProperties().queryProp("@kind");
        local = NULL == kind || 0 != stricmp("topLevelKey", kind);

        if (!local)
            width--; // 1 part == No n distributed / Monolithic key
        if (width > container.queryJob().querySlaves())
            throw MakeActivityException(this, 0, "Unsupported: keydiff(%s, %s) - Cannot diff a key that's wider(%d) than the target cluster size(%d)", originalIndexFile->queryLogicalName(), newIndexFile->queryLogicalName(), width, container.queryJob().querySlaves());

        queryThorFileManager().noteFileRead(container.queryJob(), originalIndexFile);
        queryThorFileManager().noteFileRead(container.queryJob(), newIndexFile);

        IArrayOf<IGroup> groups;
        OwnedRoxieString outputName(helper->getOutputName());
        fillClusterArray(container.queryJob(), outputName, clusters, groups);
        patchDesc.setown(queryThorFileManager().create(container.queryJob(), outputName, clusters, groups, 0 != (KDPoverwrite & helper->getFlags()), 0, !local, width));
    }

    virtual void init()
    {
        HashDistributeMasterBase::init();

        // JCSMORE should common up some with indexread
        IHThorKeyedDistributeArg *helper = (IHThorKeyedDistributeArg *)queryHelper();

        StringBuffer scoped;
        OwnedRoxieString indexFileName(helper->getIndexFileName());
        queryThorFileManager().addScope(container.queryJob(), indexFileName, scoped);
        file.setown(queryThorFileManager().lookup(container.queryJob(), indexFileName));
        if (!file)
            throw MakeActivityException(this, 0, "KeyedDistribute: Failed to find key: %s", scoped.str());
        if (0 == file->numParts())
            throw MakeActivityException(this, 0, "KeyedDistribute: Can't distribute based on an empty key: %s", scoped.str());

        checkFormatCrc(this, file, helper->getFormatCrc(), true);
        Owned<IFileDescriptor> fileDesc = file->getFileDescriptor();
        Owned<IPartDescriptor> tlkDesc = fileDesc->getPart(fileDesc->numParts()-1);
        if (!tlkDesc->queryProperties().hasProp("@kind") || 0 != stricmp("topLevelKey", tlkDesc->queryProperties().queryProp("@kind")))
            throw MakeActivityException(this, 0, "Cannot distribute using a non-distributed key: '%s'", scoped.str());
        unsigned location;
        OwnedIFile iFile;
        StringBuffer filePath;
        if (!getBestFilePart(this, *tlkDesc, iFile, location, filePath))
            throw MakeThorException(TE_FileNotFound, "Top level key part does not exist, for key: %s", file->queryLogicalName());
        OwnedIFileIO iFileIO = iFile->open(IFOread);
        assertex(iFileIO);

        tlkMb.append(iFileIO->size());
        ::read(iFileIO, 0, (size32_t)iFileIO->size(), tlkMb);

        queryThorFileManager().noteFileRead(container.queryJob(), file);
    }

        const void *nextRow()
        {
            if (eoi || activity.abortSoon)
                return false;

            try
            {
                while (xmlParser->next()) {
                    if (lastMatch)
                    {
                        RtlDynamicRowBuilder row(allocator);
                        size32_t sz = xmlTransformer->transform(row, lastMatch, this);
                        lastMatch.clear();
                        if (sz) {
                            localOffset = 0;
                            ++progress;
                            return row.finalizeRowClear(sz);
                        }
                    }
                }
            }
            catch (IXMLReadException *e)
            {
                if (XmlRead_syntax != e->errorCode())
                    throw;
                Owned<IException> _e = e;
                offset_t localFPos = makeLocalFposOffset(activity.queryContainer().queryJob().queryMyRank()-1, e->queryOffset());
                StringBuffer context;
                context.append("Logical filename = ").append(activity.logicalFilename).newline();
                context.append("Local fileposition = ");
                _WINREV8(localFPos);
                context.append("0x");
                appendDataAsHex(context, sizeof(localFPos), &localFPos);
                context.newline();
                context.append(e->queryContext());
                throw createXmlReadException(e->errorCode(), e->queryDescription(), context.str(), e->queryLine(), e->queryOffset());
            }
            catch (IOutOfMemException *e)
            {
                StringBuffer s("XMLRead actId(");
                s.append(activity.queryContainer().queryId()).append(") out of memory.").newline();
                s.append("INTERNAL ERROR ").append(e->errorCode());
                Owned<IException> e2 = MakeActivityException(&activity, e, "%s", s.str());
                e->Release();
                throw e2.getClear();
            }
            catch (IException *e)
            {
                StringBuffer s("XMLRead actId(");
                s.append(activity.queryContainer().queryId());
                s.append(") INTERNAL ERROR ").append(e->errorCode());
                Owned<IException> e2 = MakeActivityException(&activity, e, "%s", s.str());
                e->Release();
                throw e2.getClear();
            }
            eoi = true;
            return NULL;
        }

 CATCH_NEXTROW()
 {
     ActivityTimer t(totalCycles, timeActivities, NULL);
     if(abortSoon || eof || eogNext)
     {
         eogNext = false;
         return NULL;
     }
     if (refill) {
         refill=false;
         index=0;
         try
         {
             group.reset(false);
             loop {
                 OwnedConstThorRow row = input->nextRow();
                 if (!row)
                     break;
                 group.append(row.getClear());
                 if (group.isFull()) {
                     StringBuffer errStr("GROUPSORT");
                     errStr.append("(").append(container.queryId()).append(") ");
                     errStr.append("exceeded available memory. records=").append(group.ordinality()).append(", memory usage=").append((unsigned)(group.totalSize()/1024)).append('k');
                     IException *e = MakeActivityException(this, TE_TooMuchData, "%s", errStr.str());
                     EXCLOG(e, NULL);
                     throw e;
                 }
             }
             if (group.ordinality()==0) {
                 eof = true;
                 return NULL;
             }
             group.sort(*icompare,!unstable);
         }
         catch (IOutOfMemException *e)
         {
             StringBuffer errStr("GROUPSORT");
             errStr.append("(").append(container.queryId()).append(") ");
             errStr.append("exceeded available memory. records=").append(group.ordinality()).append(", memory usage=").append((unsigned)(group.totalSize()/1024)).append('k');
             errStr.append(": ").append(e->errorCode()).append(", ");
             e->errorMessage(errStr);
             e->Release();
             IException *e2 = MakeActivityException(this, TE_TooMuchData, "%s", errStr.str());
             EXCLOG(e2, NULL);
             throw e2;
         }
     }
     if(index >= group.ordinality())
     {
         refill = true;
         return NULL;    // eog
     }
     const void *row = group.itemClear(index++);
     assertex(row);
     dataLinkIncrement();
     return row;
 }

    virtual void init()
    {
        OwnedRoxieString fname(helper->getFileName());
        dlfn.set(fname);
        isLocal = 0 != (TIWlocal & helper->getFlags());
        unsigned maxSize = helper->queryDiskRecordSize()->getMinRecordSize();
        if (maxSize > KEYBUILD_MAXLENGTH)
            throw MakeActivityException(this, 0, "Index minimum record length (%d) exceeds %d internal limit", maxSize, KEYBUILD_MAXLENGTH);

        singlePartKey = 0 != (helper->getFlags() & TIWsmall) || dlfn.isExternal();
        clusters.kill();
        unsigned idx=0;
        while (true)
        {
            OwnedRoxieString cluster(helper->getCluster(idx));
            if(!cluster)
                break;
            clusters.append(cluster);
            idx++;
        }
        IArrayOf<IGroup> groups;
        if (singlePartKey)
        {
            isLocal = true;
            buildTlk = false;
        }
        else if (!isLocal || globals->getPropBool("@buildLocalTlks", true))
            buildTlk = true;

        fillClusterArray(container.queryJob(), fname, clusters, groups);
        unsigned restrictedWidth = 0;
        if (TIWhaswidth & helper->getFlags())
        {
            restrictedWidth = helper->getWidth();
            if (restrictedWidth > container.queryJob().querySlaves())
                throw MakeActivityException(this, 0, "Unsupported, can't refactor to width(%d) larger than host cluster(%d)", restrictedWidth, container.queryJob().querySlaves());
            else if (restrictedWidth < container.queryJob().querySlaves())
            {
                if (!isLocal)
                    throw MakeActivityException(this, 0, "Unsupported, refactoring to few parts only supported for local indexes.");
                assertex(!singlePartKey);
                unsigned gwidth = groups.item(0).ordinality();
                if (0 != container.queryJob().querySlaves() % gwidth)
                    throw MakeActivityException(this, 0, "Unsupported, refactored target size (%d) must be factor of thor cluster width (%d)", groups.item(0).ordinality(), container.queryJob().querySlaves());
                if (0 == restrictedWidth)
                    restrictedWidth = gwidth;
                ForEachItemIn(g, groups)
                {
                    IGroup &group = groups.item(g);
                    if (gwidth != groups.item(g).ordinality())
                        throw MakeActivityException(this, 0, "Unsupported, cannot output multiple refactored widths, targeting cluster '%s' and '%s'", clusters.item(0), clusters.item(g));
                    if (gwidth != restrictedWidth)
                        groups.replace(*group.subset((unsigned)0, restrictedWidth), g);
                }
                refactor = true;
            }

void CDiskPartHandlerBase::open() 
{
    unsigned location;
    StringBuffer filePath;
    if (!(globals->getPropBool("@autoCopyBackup", true)?ensurePrimary(&activity, *partDesc, iFile, location, filePath):getBestFilePart(&activity, *partDesc, iFile, location, filePath, &activity)))
    {
        StringBuffer locations;
        IException *e = MakeActivityException(&activity, TE_FileNotFound, "No physical file part for logical file %s, found at given locations: %s (Error = %d)", activity.logicalFilename.get(), getFilePartLocations(*partDesc, locations).str(), GetLastError());
        EXCLOG(e, NULL);
        throw e;
    }
    filename.set(iFile->queryFilename());
    ActPrintLog(&activity, "%s[part=%d]: reading physical file '%s' (logical file = %s)", kindStr, which, filePath.str(), activity.logicalFilename.get());
    if (checkFileCrc)
    {
        CDateTime createTime, modifiedTime, accessedTime;
        iFile->getTime(&createTime, &modifiedTime, &accessedTime);
        const char *descModTimeStr = partDesc->queryProperties().queryProp("@modified");
        CDateTime descModTime;
        descModTime.setString(descModTimeStr);
        if (!descModTime.equals(modifiedTime, false))
        {
            StringBuffer diskTimeStr;
            ActPrintLog(&activity, "WARNING: file (%s); modified date stamps on disk (%s) are not equal to published modified data (%s)", filePath.str(), modifiedTime.getString(diskTimeStr).str(), descModTimeStr);
        }
    }

    ActPrintLog(&activity, "%s[part=%d]: Base offset to %"I64F"d", kindStr, which, fileBaseOffset);

    if (compressed)
    {
        ActPrintLog(&activity, "Reading %s compressed file: %s", (NULL != activity.eexp.get())?"encrypted":blockCompressed?"block":"row", filename.get());
        if (checkFileCrc)
        {
            checkFileCrc = false;
            if (activity.crcCheckCompressed) // applies to encrypted too, (optional, default off)
            {
                ActPrintLog(&activity, "Calculating crc for file: %s", filename.get());
                unsigned calcCrc = iFile->getCRC();
                // NB: for compressed files should always be ~0
                ActPrintLog(&activity, "Calculated crc = %x, storedCrc = %x", calcCrc, storedCrc);
                if (calcCrc != storedCrc)
                {
                    IThorException *e = MakeActivityException(&activity, TE_FileCrc, "CRC Failure validating compressed file: %s", iFile->queryFilename());
                    e->setAudience(MSGAUD_operator);
                    throw e;
                }
            }
        }
    }
}

    void process()
    {
        CMessageBuffer msg;
        unsigned inputs = container.getInputs();
        unsigned slaves = container.queryJob().querySlaves();
        unsigned s;
        bool readSome=false, slaveReadSome;

        IntArray replyTags;
        for (s=0; s<slaves; s++)
            replyTags.append(0);
        while (inputs>1)
        {
            inputs--;
            for (s=0; s<slaves; s++)
            {
                rank_t sender;
                if (!receiveMsg(msg, RANK_ALL, replyTag, &sender)) return;

                replyTags.replace(msg.getReplyTag(), ((int)sender)-1);
                msg.read(slaveReadSome);
                if (slaveReadSome) readSome = true;
            }
            msg.clear().append(readSome);
            for (s=0; s<slaves; s++)
            {
                if (!queryJobChannel().queryJobComm().send(msg, ((rank_t) s+1), (mptag_t) replyTags.item(s), LONGTIMEOUT))
                    throw MakeActivityException(this, 0, "Failed to give result to slave");
            }
            if (readSome) // got some, have told slaves to ignore rest, so finish
                break;
        }
    }

void cancelReplicates(CActivityBase *activity, IPartDescriptor &partDesc)
{
    RemoteFilename rfn;
    IFileDescriptor &fileDesc = partDesc.queryOwner();
    unsigned copies = partDesc.numCopies();
    unsigned c=1;
    for (; c<copies; c++)
    {
        unsigned replicateCopy;
        unsigned clusterNum = partDesc.copyClusterNum(c, &replicateCopy);
        rfn.clear();
        partDesc.getFilename(c, rfn);
        StringBuffer dstName;
        rfn.getPath(dstName);
        assertex(dstName.length());

        if (replicateCopy>0)  
        {
            try
            {
                queryThor().queryBackup().cancel(dstName.str());
            }
            catch (IException *e)
            {
                Owned<IThorException> re = MakeActivityException(activity, e, "Error cancelling backup '%s'", dstName.str());
                ActPrintLog(&activity->queryContainer(), e, NULL);
                e->Release();
            }
        }
    }
}

    virtual void init()
    {
        CSortBaseActivityMaster::init();
        IHThorSortArg *helper = (IHThorSortArg *)queryHelper();
        IHThorAlgorithm *algo = static_cast<IHThorAlgorithm *>(helper->selectInterface(TAIalgorithm_1));
        OwnedRoxieString algoname(algo->getAlgorithm());
        unsigned flags = algo->getAlgorithmFlags();
        if (algoname && (0 != stricmp(algoname, "quicksort")))
        {
            Owned<IException> e = MakeActivityException(this, 0, "Ignoring, unsupported sort order algorithm '%s'", algoname.get());
            reportExceptionToWorkunit(container.queryJob().queryWorkUnit(), e);
        }
        OwnedRoxieString cosortlogname(helper->getSortedFilename());
        if (cosortlogname&&*cosortlogname)
        {
            Owned<IDistributedFile> coSortFile = queryThorFileManager().lookup(container.queryJob(), cosortlogname);
            addReadFile(coSortFile);
            Owned<IFileDescriptor> fileDesc = coSortFile->getFileDescriptor();
            unsigned o;
            for (o=0; o<fileDesc->numParts(); o++)
            {
                Owned<IPartDescriptor> partDesc = fileDesc->getPart(o);
                if (cosortfilenames.length())
                    cosortfilenames.append("|");

                // JCSMORE - picking the primary here, means no automatic use of backup copy, could use RMF's possibly.
                getPartFilename(*partDesc, 0, cosortfilenames);
            }
        }
    }

    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
    {
        isLocal = 0 != (TIWlocal & helper->getFlags());

        mpTag = container.queryJobChannel().deserializeMPTag(data);
        mpTag2 = container.queryJobChannel().deserializeMPTag(data);
        data.read(active);
        if (active)
        {
            data.read(logicalFilename);
            partDesc.setown(deserializePartFileDescriptor(data));
        }

        data.read(singlePartKey);
        data.read(refactor);
        if (singlePartKey)
            buildTlk = false;
        else
        {
            data.read(buildTlk);
            if (firstNode())
            {
                if (buildTlk)
                    tlkDesc.setown(deserializePartFileDescriptor(data));
                else if (!isLocal) // existing tlk then..
                {
                    tlkDesc.setown(deserializePartFileDescriptor(data));
                    unsigned c;
                    data.read(c);
                    while (c--)
                    {
                        RemoteFilename rf;
                        rf.deserialize(data);
                        if (!existingTlkIFile)
                        {
                            Owned<IFile> iFile = createIFile(rf);
                            if (iFile->exists())
                                existingTlkIFile.set(iFile);
                        }
                    }
                    if (!existingTlkIFile)
                        throw MakeActivityException(this, TE_FileNotFound, "Top level key part does not exist, for key");
                }
            }
        }

        IOutputMetaData * diskSize = helper->queryDiskRecordSize();
        assertex(!(diskSize->getMetaFlags() & MDFneedserializedisk));
        if (diskSize->isVariableSize())
        {
            if (TIWmaxlength & helper->getFlags())
                maxDiskRecordSize = helper->getMaxKeySize();
            else
                maxDiskRecordSize = KEYBUILD_MAXLENGTH; //Current default behaviour, could be improved in the future
        }
        else
            maxDiskRecordSize = diskSize->getFixedSize();
        reportOverflow = false;
    }

    void process()
    {
        ActPrintLog("ReDistributeActivityMaster::process");
        HashDistributeMasterBase::process();        
        IHThorHashDistributeArg *helper = (IHThorHashDistributeArg *)queryHelper(); 
        unsigned n = container.queryJob().querySlaves();
        MemoryAttr ma;
        offset_t *sizes = (offset_t *)ma.allocate(sizeof(offset_t)*n);
        unsigned i;
        try {
            for (i=0;i<n;i++) {
                if (abortSoon)
                    return;
                CMessageBuffer mb;
#ifdef _TRACE
                ActPrintLog("ReDistribute process, Receiving on tag %d",statstag);
#endif
                rank_t sender;
                if (!receiveMsg(mb, RANK_ALL, statstag, &sender)||abortSoon) 
                    return;
#ifdef _TRACE
                ActPrintLog("ReDistribute process, Received size from %d",sender);
#endif
                sender--;
                assertex((unsigned)sender<n);
                mb.read(sizes[sender]);
            }
            ActPrintLog("ReDistributeActivityMaster::process sizes got");
            for (i=0;i<n;i++) {
                CMessageBuffer mb;
                mb.append(n*sizeof(offset_t),sizes);
#ifdef _TRACE
                ActPrintLog("ReDistribute process, Replying to node %d tag %d",i+1,statstag);
#endif
                if (!container.queryJob().queryJobComm().send(mb, (rank_t)i+1, statstag))
                    return;
            }
            // check if any max skew broken
            double maxskew = helper->getTargetSkew();
            if (maxskew>helper->getSkew()) {
                offset_t tot = 0;
                for (i=0;i<n;i++) 
                    tot += sizes[i];
                offset_t avg = tot/n;
                for (i=0;i<n;i++) {
                    double r = ((double)sizes[i]-(double)avg)/(double)avg;
                    if ((r>=maxskew)||(-r>maxskew)) {
                        throw MakeActivityException(this, TE_DistributeFailedSkewExceeded, "DISTRIBUTE maximum skew exceeded (node %d has %"I64F"d, average is %"I64F"d)",i+1,sizes[i],avg);
                    }               
                }
            }

        }
        catch (IException *e) {
            ActPrintLog(e,"ReDistribute");
            throw;
        }
        ActPrintLog("ReDistributeActivityMaster::process exit");
    }

    virtual void init()
    {
        CMasterActivity::init();
        helper = (IHThorKeyPatchArg *)queryHelper();
        OwnedRoxieString originalHelperName(helper->getOriginalName());
        OwnedRoxieString patchHelperName(helper->getPatchName());
        OwnedRoxieString outputHelperName(helper->getOutputName());
        StringBuffer expandedFileName;
        queryThorFileManager().addScope(container.queryJob(), originalHelperName, expandedFileName, false);
        originalName.set(expandedFileName);
        queryThorFileManager().addScope(container.queryJob(), patchHelperName, expandedFileName.clear(), false);
        patchName.set(expandedFileName);
        queryThorFileManager().addScope(container.queryJob(), outputHelperName, expandedFileName.clear(), false);
        outputName.set(expandedFileName);

        originalIndexFile.setown(queryThorFileManager().lookup(container.queryJob(), originalHelperName));
        patchFile.setown(queryThorFileManager().lookup(container.queryJob(), patchHelperName));
        
        if (originalIndexFile->numParts() != patchFile->numParts())
            throw MakeActivityException(this, TE_KeyPatchIndexSizeMismatch, "Index %s and patch %s differ in width", originalName.get(), patchName.get());
        if (originalIndexFile->querySuperFile() || patchFile->querySuperFile())
            throw MakeActivityException(this, 0, "Patching super files not supported");
        
        addReadFile(originalIndexFile);
        addReadFile(patchFile);

        width = originalIndexFile->numParts();

        originalDesc.setown(originalIndexFile->getFileDescriptor());
        patchDesc.setown(patchFile->getFileDescriptor());

        Owned<IPartDescriptor> tlkDesc = originalDesc->getPart(originalDesc->numParts()-1);
        const char *kind = tlkDesc->queryProperties().queryProp("@kind");
        local = NULL == kind || 0 != stricmp("topLevelKey", kind);

        if (!local && width > 1)
            width--; // 1 part == No n distributed / Monolithic key
        if (width > container.queryJob().querySlaves())
            throw MakeActivityException(this, 0, "Unsupported: keypatch(%s, %s) - Cannot patch a key that's wider(%d) than the target cluster size(%d)", originalIndexFile->queryLogicalName(), patchFile->queryLogicalName(), width, container.queryJob().querySlaves());

        IArrayOf<IGroup> groups;
        fillClusterArray(container.queryJob(), outputName, clusters, groups);
        newIndexDesc.setown(queryThorFileManager().create(container.queryJob(), outputName, clusters, groups, 0 != (KDPoverwrite & helper->getFlags()), 0, !local, width));
        if (!local)
            newIndexDesc->queryPart(newIndexDesc->numParts()-1)->queryProperties().setProp("@kind", "topLevelKey");
    }

 void init()
 {
     workunitWriteLimit = getOptInt(THOROPT_OUTPUTLIMIT, DEFAULT_WUIDWRITE_LIMIT);
     if (workunitWriteLimit>DALI_RESULT_OUTPUTMAX)
         throw MakeActivityException(this, 0, "Dali result outputs are restricted to a maximum of %d MB, the default limit is %d MB. A huge dali result usually indicates the ECL needs altering.", DALI_RESULT_OUTPUTMAX, DEFAULT_WUIDWRITE_LIMIT);
     assertex(workunitWriteLimit<=0x1000); // 32bit limit because MemoryBuffer/CMessageBuffers involved etc.
     workunitWriteLimit *= 0x100000;
 }

void CDiskReadMasterBase::init()
{
    IHThorDiskReadBaseArg *helper = (IHThorDiskReadBaseArg *) queryHelper();
    fileName.setown(helper->getFileName());
    file.setown(queryThorFileManager().lookup(container.queryJob(), fileName, 0 != ((TDXtemporary|TDXjobtemp) & helper->getFlags()), 0 != (TDRoptional & helper->getFlags()), true));

    if (file)
    {
        if (file->numParts() > 1)
            fileDesc.setown(getConfiguredFileDescriptor(*file));
        else
            fileDesc.setown(file->getFileDescriptor());
        reInit = 0 != (helper->getFlags() & (TDXvarfilename|TDXdynamicfilename));
        if (container.queryLocal() || helper->canMatchAny()) // if local, assume may match
        {
            bool local;
            if (0 == (TDXtemporary & helper->getFlags())) // don't add temp files
            {
                queryThorFileManager().noteFileRead(container.queryJob(), file);
                local = container.queryLocal();
            }
            else
                local = false;
            mapping.setown(getFileSlaveMaps(file->queryLogicalName(), *fileDesc, container.queryJob().queryUserDescriptor(), container.queryJob().querySlaveGroup(), local, false, hash, file->querySuperFile()));
        }
        if (0 != (helper->getFlags() & TDRfilenamecallback)) // only get/serialize if using virtual file name fields
        {
            IDistributedSuperFile *super = file->querySuperFile();
            if (super)
            {
                unsigned numsubs = super->numSubFiles(true);
                unsigned s=0;
                for (; s<numsubs; s++)
                {
                    IDistributedFile &subfile = super->querySubFile(s, true);
                    subfileLogicalFilenames.append(subfile.queryLogicalName());
                }
            }
        }
        validateFile(file);
        void *ekey;
        size32_t ekeylen;
        helper->getEncryptKey(ekeylen,ekey);
        bool encrypted = fileDesc->queryProperties().getPropBool("@encrypted");
        if (0 != ekeylen)
        {
            memset(ekey,0,ekeylen);
            free(ekey);
            if (!encrypted)
            {
                Owned<IException> e = MakeActivityWarning(&container, TE_EncryptionMismatch, "Ignoring encryption key provided as file '%s' was not published as encrypted", fileName.get());
                container.queryJob().fireException(e);
            }
        }
        else if (encrypted)
            throw MakeActivityException(this, 0, "File '%s' was published as encrypted but no encryption key provided", fileName.get());
    }
}

 virtual void init()
 {
     CMasterActivity::init();
     workunitWriteLimit = activityMaxSize ? activityMaxSize : getOptInt(THOROPT_OUTPUTLIMIT, DEFAULT_WUIDWRITE_LIMIT);
     if (workunitWriteLimit>DALI_RESULT_OUTPUTMAX)
         throw MakeActivityException(this, 0, "Configured max result size, %d MB, exceeds absolute max limit of %d MB. A huge Dali result usually indicates the ECL needs altering.", workunitWriteLimit, DALI_RESULT_OUTPUTMAX);
     assertex(workunitWriteLimit<=0x1000); // 32bit limit because MemoryBuffer/CMessageBuffers involved etc.
     workunitWriteLimit *= 0x100000;
 }

 virtual void init()
 {
     CMasterActivity::init();
     // In absense of OPT_OUTPUTLIMIT check pre 5.2 legacy name OPT_OUTPUTLIMIT_LEGACY
     workunitWriteLimit = activityMaxSize ? activityMaxSize : getOptInt(OPT_OUTPUTLIMIT, getOptInt(OPT_OUTPUTLIMIT_LEGACY, defaultDaliResultLimit));
     if (workunitWriteLimit>defaultDaliResultOutputMax)
         throw MakeActivityException(this, 0, "Configured max result size, %d MB, exceeds absolute max limit of %d MB. A huge Dali result usually indicates the ECL needs altering.", workunitWriteLimit, defaultDaliResultOutputMax);
     assertex(workunitWriteLimit<=0x1000); // 32bit limit because MemoryBuffer/CMessageBuffers involved etc.
     workunitWriteLimit *= 0x100000;
 }

 void preStart(size32_t parentExtractSz, const byte *parentExtract)
 {
     CMasterActivity::preStart(parentExtractSz, parentExtract);
     IHThorKeyPatchArg *helper = (IHThorKeyPatchArg *) queryHelper();
     if (0==(KDPoverwrite & helper->getFlags()))
     {
         if (KDPvaroutputname & helper->getFlags()) return;
         Owned<IDistributedFile> file = queryThorFileManager().lookup(container.queryJob(), outputName, false, true);
         if (file)
             throw MakeActivityException(this, TE_OverwriteNotSpecified, "Cannot write %s, file already exists (missing OVERWRITE attribute?)", file->queryLogicalName());
     }
 }

 virtual void init()
 {
     IHThorSortArg *helper = (IHThorSortArg *)queryHelper();
     IHThorAlgorithm *algo = static_cast<IHThorAlgorithm *>(helper->selectInterface(TAIalgorithm_1));
     char const *algoname = algo->queryAlgorithm();
     unsigned flags = algo->getAlgorithmFlags();
     if (algoname && (0 != stricmp(algoname, "quicksort")))
     {
         Owned<IException> e = MakeActivityException(this, 0, "Ignoring, unsupported sort order algorithm '%s'", algoname);
         reportExceptionToWorkunit(container.queryJob().queryWorkUnit(), e);
     }
 }

 CATCH_NEXTROW()
 {
     ActivityTimer t(totalCycles, timeActivities, NULL);
     OwnedConstThorRow ret = input->nextRow();
     if (ret && prev && icompare->docompare(prev, ret) > 0)
     {
         // MORE - better to give mismatching rows than indexes?
         throw MakeActivityException(this, TE_NotSorted, "detected incorrectly sorted rows (row %"RCPF"d,  %"RCPF"d))", getDataLinkCount(), getDataLinkCount()+1);
     }
     prev.set(ret);
     if (ret)
         dataLinkIncrement();
     return ret.getClear();
 }