Merge branch 'candidate-7.8.x' into candidate-7.10.x

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

esp/bindings/http/platform/httpbinding.cpp

@@ -753,14 +753,29 @@ bool EspHttpBinding::basicAuth(IEspContext* ctx)
     bool authenticated = m_secmgr->authorize(*user, rlist, ctx->querySecureContext());
     if(!authenticated)
     {
-        const char *desc = nullptr;
-        if (user->getAuthenticateStatus() == AS_PASSWORD_EXPIRED || user->getAuthenticateStatus() == AS_PASSWORD_VALID_BUT_EXPIRED)
-            desc = "ESP password is expired";
-        else
-            desc = "Access Denied: User or password invalid";
-        ctx->AuditMessage(AUDIT_TYPE_ACCESS_FAILURE, "Authentication", desc);
+        VStringBuffer err("User %s : ", user->getName());
+        switch (user->getAuthenticateStatus())
+        {
+        case AS_PASSWORD_EXPIRED :
+        case AS_PASSWORD_VALID_BUT_EXPIRED :
+            err.append("Password expired");
+            break;
+        case AS_ACCOUNT_DISABLED :
+            err.append("Account disabled");
+            break;
+        case AS_ACCOUNT_EXPIRED :
+            err.append("Account expired");
+            break;
+        case AS_ACCOUNT_LOCKED :
+            err.append("Account locked");
+            break;
+        case AS_INVALID_CREDENTIALS :
+        default:
+            err.append("Access Denied: User or password invalid");
+        }
+        ctx->AuditMessage(AUDIT_TYPE_ACCESS_FAILURE, "Authentication", err.str());
         ctx->setAuthError(EspAuthErrorNotAuthenticated);
-        ctx->setRespMsg(desc);
+        ctx->setRespMsg(err.str());
         return false;
     }
     bool authorized = true;

esp/bindings/http/platform/httpservice.cpp

@@ -1069,6 +1069,7 @@ EspAuthState CEspHttpServer::preCheckAuth(EspAuthRequest& authReq)
                 clearCookie(authReq.authBinding->querySessionIDCookieName());
                 clearCookie(SESSION_ID_TEMP_COOKIE);
                 clearCookie(SESSION_TIMEOUT_COOKIE);
+                clearCookie(USER_ACCT_ERROR_COOKIE);
             }
             else
                 clearSessionCookies(authReq);
@@ -1205,10 +1206,11 @@ void CEspHttpServer::verifyCookie(EspAuthRequest& authReq, CESPCookieVerificatio
         verifyESPAuthenticatedCookie(authReq, cookie);
     else if (strieq(name, USER_NAME_COOKIE))
         verifyESPUserNameCookie(authReq, cookie);
-    else if (strieq(name, SESSION_TIMEOUT_COOKIE) || strieq(name, SESSION_AUTH_MSG_COOKIE))
+    else if (strieq(name, SESSION_TIMEOUT_COOKIE) || strieq(name, SESSION_AUTH_MSG_COOKIE) || strieq(name, USER_ACCT_ERROR_COOKIE))
     {
         //SESSION_TIMEOUT_COOKIE: used to pass timeout settings to a client.
         //SESSION_AUTH_MSG_COOKIE: used to pass authentication message to a client.
+        //USER_ACCT_ERROR_COOKIE: used to pass user account status to a client.
         //A client should clean it as soon as received. ESP always returns invalid if it is asked.
         cookie.verificationDetails.set("ESP cannot verify this cookie. It is one-time use only.");
     }
@@ -2050,18 +2052,34 @@ void CEspHttpServer::logoutSession(EspAuthRequest& authReq, unsigned sessionID,
 EspAuthState CEspHttpServer::handleAuthFailed(bool sessionAuth, EspAuthRequest& authReq, bool unlock, const char* msg)
 {
     ISecUser *user = authReq.ctx->queryUser();
-    if (user && user->getAuthenticateStatus() == AS_PASSWORD_VALID_BUT_EXPIRED)
+    if (user)
     {
-        ESPLOG(LogMin, "ESP password expired for %s. Asking update ...", authReq.ctx->queryUserId());
-        if (sessionAuth) //For session auth, store the userid to cookie for the updatepasswordinput form.
-            addCookie(SESSION_ID_TEMP_COOKIE, authReq.ctx->queryUserId(), 0, true);
-        m_response->redirect(*m_request.get(), "/esp/updatepasswordinput");
-        return authSucceeded;
+        switch (user->getAuthenticateStatus())
+        {
+        case AS_PASSWORD_VALID_BUT_EXPIRED :
+            ESPLOG(LogMin, "ESP password expired for %s. Asking update ...", authReq.ctx->queryUserId());
+            if (sessionAuth) //For session auth, store the userid to cookie for the updatepasswordinput form.
+                addCookie(SESSION_ID_TEMP_COOKIE, authReq.ctx->queryUserId(), 0, true);
+            m_response->redirect(*m_request.get(), "/esp/updatepasswordinput");
+            return authSucceeded;
+        case AS_PASSWORD_EXPIRED :
+            ESPLOG(LogMin, "ESP password expired for %s", authReq.ctx->queryUserId());
+            break;
+        case AS_ACCOUNT_DISABLED :
+            ESPLOG(LogMin, "Account disabled for %s", authReq.ctx->queryUserId());
+            addCookie(USER_ACCT_ERROR_COOKIE, "Account Disabled", 0, false);
+            break;
+        case AS_ACCOUNT_EXPIRED :
+            ESPLOG(LogMin, "Account expired for %s", authReq.ctx->queryUserId());
+            addCookie(USER_ACCT_ERROR_COOKIE, "Account Expired", 0, false);
+            break;
+        case AS_ACCOUNT_LOCKED :
+            ESPLOG(LogMin, "Account locked for %s", authReq.ctx->queryUserId());
+            addCookie(USER_ACCT_ERROR_COOKIE, "Account Locked", 0, false);
+            break;
+        }
     }
 
-    if (user && (user->getAuthenticateStatus() == AS_PASSWORD_EXPIRED))
-        ESPLOG(LogMin, "ESP password expired for %s", authReq.ctx->queryUserId());
-
     if (unlock)
     {
         ESPLOG(LogMin, "Unlock failed: invalid user name or password.");
@@ -2228,6 +2246,7 @@ void CEspHttpServer::clearSessionCookies(EspAuthRequest& authReq)
     clearCookie(SESSION_AUTH_OK_COOKIE);
     clearCookie(SESSION_AUTH_MSG_COOKIE);
     clearCookie(SESSION_TIMEOUT_COOKIE);
+    clearCookie(USER_ACCT_ERROR_COOKIE);
 }
 
 void CEspHttpServer::clearCookie(const char* cookieName)

esp/platform/espcontext.hpp

@@ -43,6 +43,7 @@ static const char* const SESSION_TIMEOUT_COOKIE = "ESPSessionTimeoutSeconds";
 static const char* const SESSION_ID_TEMP_COOKIE = "ESPAuthIDTemp";
 static const char* const SESSION_AUTH_OK_COOKIE = "ESPAuthenticated";
 static const char* const SESSION_AUTH_MSG_COOKIE = "ESPAuthenticationMSG";
+static const char* const USER_ACCT_ERROR_COOKIE = "ESPUserAcctError";
 static const char* const DEFAULT_LOGIN_URL = "/esp/files/Login.html";
 static const char* const DEFAULT_LOGIN_LOGO_URL = "/esp/files/eclwatch/img/Loginlogo.png";
 static const char* const DEFAULT_GET_USER_NAME_URL = "/esp/files/GetUserName.html";

esp/scm/ws_account.ecm

@@ -45,6 +45,7 @@ ESPresponse [exceptions_inline] MyAccountResponse
     [min_ver("1.04")] string accountType; //"User" or "Administrator"
     [min_ver("1.04")] bool passwordNeverExpires;
     [min_ver("1.04")] bool passwordIsExpired;
+    [min_ver("1.05")] int accountStatus;//disabled, expired, locked, etc. See 'enum authStatus' in seclib.hpp
 };
 
 
@@ -74,7 +75,7 @@ ESPresponse [exceptions_inline] VerifyUserResponse
 };
 
 //Kevin/russ does this method need feature level check?
-ESPservice [auth_feature("NONE"), version("1.04"), default_client_version("1.04"), exceptions_inline("./smc_xslt/exceptions.xslt")] ws_account
+ESPservice [auth_feature("NONE"), version("1.05"), default_client_version("1.05"), exceptions_inline("./smc_xslt/exceptions.xslt")] ws_account
 {
     ESPmethod [client_xslt("/esp/xslt/account_myaccount.xslt")] MyAccount(MyAccountRequest, MyAccountResponse);
     ESPmethod [client_xslt("/esp/xslt/account_input.xslt")] UpdateUserInput(UpdateUserInputRequest, UpdateUserInputResponse);

esp/services/ws_account/ws_accountService.cpp

@@ -206,6 +206,11 @@ bool Cws_accountEx::onMyAccount(IEspContext &context, IEspMyAccountRequest &req,
                 resp.setAccountType(secmgr->isSuperUser(user) ? "Administrator" : "User");
                 resp.setPasswordIsExpired(userInContext->getAuthenticateStatus() == AS_PASSWORD_EXPIRED || userInContext->getAuthenticateStatus() == AS_PASSWORD_VALID_BUT_EXPIRED);
             }
+
+            if (version >= 1.05)
+            {
+                resp.setAccountStatus(userInContext->getAuthenticateStatus());
+            }
         }
     }
     catch(IException* e)

esp/services/ws_machine/ws_machineService.cpp

@@ -3443,9 +3443,12 @@ void CUsageCacheReader::addDropZoneUsageReq(IConstEnvironment* constEnv, const c
         throw MakeStringException(ECLWATCH_INVALID_INPUT, "Empty DropZone name");
 
     Owned<IConstDropZoneInfo> envDropZone = constEnv->getDropZone(name);
-    if (!envDropZone || !envDropZone->isECLWatchVisible())
+    if (!envDropZone)
         throw MakeStringException(ECLWATCH_INVALID_INPUT, "Dropzone %s not found", name);
 
+    if (!envDropZone->isECLWatchVisible())
+        return;
+
     SCMStringBuffer directory;
     envDropZone->getDirectory(directory);
     if (directory.length() == 0)

fs/dafsclient/rmtclient.cpp

@@ -388,16 +388,16 @@ void flushDaFsSocket(ISocket *socket)
     try
     {
         sendDaFsBuffer(socket, sendbuf);
-        char buf[1024];
+        char buf[16*1024];
         for (;;)
         {
-            Sleep(1000);    // breathe
             size32_t szread;
-            SOCKREADTMS(socket)(buf, 1, sizeof(buf), szread, 1000*60);
+            SOCKREADTMS(socket)(buf, 1, sizeof(buf), szread, 1000*30);
             totread += szread;
         }
     }
-    catch (IJSOCK_Exception *e) {
+    catch (IJSOCK_Exception *e)
+    {
         if (totread)
             PROGLOG("%d bytes discarded",totread);
         if (e->errorCode()!=JSOCKERR_timeout_expired)
@@ -411,17 +411,21 @@ void receiveDaFsBuffer(ISocket * socket, MemoryBuffer & tgt, unsigned numtries,
 {
     sRFTM tm(dafsMaxReceiveTimeMs);
     size32_t gotLength = receiveDaFsBufferSize(socket, numtries,tm.timemon);
-    if (gotLength) {
+    Owned<IException> exc;
+    if (gotLength)
+    {
         size32_t origlen = tgt.length();
-        try {
-            if (gotLength>maxsz) {
+        try
+        {
+            if (gotLength>maxsz)
+            {
                 StringBuffer msg;
                 msg.appendf("receiveBuffer maximum block size exceeded %d/%d",gotLength,maxsz);
-                PrintStackReport();
                 throw createDafsException(DAFSERR_protocol_failure,msg.str());
             }
             unsigned timeout = SERVER_TIMEOUT*(numtries?numtries:1);
-            if (tm.timemon) {
+            if (tm.timemon)
+            {
                 unsigned remaining;
                 if (tm.timemon->timedout(&remaining)||(remaining<10))
                     remaining = 10;
@@ -431,29 +435,27 @@ void receiveDaFsBuffer(ISocket * socket, MemoryBuffer & tgt, unsigned numtries,
             size32_t szread;
             SOCKREADTMS(socket)((gotLength<4000)?tgt.reserve(gotLength):tgt.reserveTruncate(gotLength), gotLength, gotLength, szread, timeout);
         }
-        catch (IJSOCK_Exception *e) {
-            if (e->errorCode()!=JSOCKERR_timeout_expired) {
-                EXCLOG(e,"receiveDaFsBuffer(1)");
-                PrintStackReport();
+        catch (IException *e)
+        {
+            exc.setown(e);
+        }
+
+        if (exc.get())
+        {
+            tgt.setLength(origlen);
+            EXCLOG(exc, "receiveDaFsBuffer");
+            PrintStackReport();
+            if (JSOCKERR_timeout_expired != exc->errorCode())
+            {
                 if (!tm.timemon||!tm.timemon->timedout())
                     flushDaFsSocket(socket);
             }
-            else {
-                EXCLOG(e,"receiveDaFsBuffer");
-                PrintStackReport();
-            }
-            tgt.setLength(origlen);
-            throw;
-        }
-        catch (IException *e) {
-            EXCLOG(e,"receiveDaFsBuffer(2)");
-            PrintStackReport();
-            if (!tm.timemon||!tm.timemon->timedout())
-                flushDaFsSocket(socket);
-            tgt.setLength(origlen);
-            throw;
+            IJSOCK_Exception *JSexc = dynamic_cast<IJSOCK_Exception *>(exc.get());
+            if (JSexc != nullptr)
+                throw LINK(JSexc);
+            else
+                throw exc.getClear();
         }
-
     }
     tgt.setEndian(__BIG_ENDIAN);
 }

rtl/eclrtl/rtlds.cpp

@@ -1881,9 +1881,23 @@ byte * MemoryBufferBuilder::ensureCapacity(size32_t required, const char * field
     dbgassertex(buffer);
     if (required > reserved)
     {
-        void * next = buffer->reserve(required-reserved);
-        self = (byte *)next - reserved;
-        reserved = required;
+        try
+        {
+            void * next = buffer->reserve(required-reserved);
+            self = (byte *)next - reserved;
+            reserved = required;
+        }
+        catch (IException *E)
+        {
+            VStringBuffer s("While allocating %u bytes for field %s", (unsigned) required, fieldName);
+            EXCLOG(E, s.str());
+            throw;
+        }
+        catch (...)
+        {
+            DBGLOG("Unknown exception while allocating %u bytes for field %s", (unsigned) required, fieldName);
+            throw;
+        }
     }
     return self;
 }

rtl/eclrtl/rtldynfield.cpp

@@ -1443,9 +1443,7 @@ private:
             }
             else
             {
-                // Note - ifblocks make this assertion invalid. We do not account for potentially omitted fields
-                // when estimating target record size.
-                if (!destRecInfo.getNumIfBlocks() && !hasBlobs)
+                if (!hasBlobs)
                     assert(offset-origOffset > estimate);  // Estimate is always supposed to be conservative
     #ifdef TRACE_TRANSLATION
                 DBGLOG("Wrote %u bytes to record (estimate was %u)\n", offset-origOffset, estimate);
@@ -1462,7 +1460,7 @@ private:
     const RtlRecord &sourceRecInfo;
     bool binarySource = true;
     type_vals callbackRawType;
-    unsigned fixedDelta = 0;  // total size of all fixed-size source fields that are not matched
+    int fixedDelta = 0;  // total size difference from all fixed size mappings
     UnsignedArray allUnmatched;  // List of all source fields that are unmatched (so that we can trace them)
     UnsignedArray variableUnmatched;  // List of all variable-size source fields that are unmatched
     FieldMatchType matchFlags = match_perfect;
@@ -1593,8 +1591,11 @@ private:
             {
                 const byte * initializer = (const byte *) field->initializer;
                 info.matchType = isVirtualInitializer(initializer) ? match_virtual : match_none;
-                size32_t defaultSize = (initializer && !isVirtualInitializer(initializer)) ? type->size(initializer, nullptr) : type->getMinSize();
-                fixedDelta -= defaultSize;
+                if ((field->flags & RFTMinifblock) == 0)
+                {
+                    size32_t defaultSize = (initializer && !isVirtualInitializer(initializer)) ? type->size(initializer, nullptr) : type->getMinSize();
+                    fixedDelta -= defaultSize;
+                }
                 if ((field->flags & RFTMispayloadfield) == 0)
                     matchFlags |= match_keychange;
                 defaulted++;
@@ -1604,6 +1605,8 @@ private:
             {
                 bool deblob = false;
                 const RtlTypeInfo *sourceType = sourceRecInfo.queryType(info.matchIdx);
+                unsigned sourceFlags = sourceRecInfo.queryField(info.matchIdx)->flags;
+                unsigned destFlags = field->flags;
                 if (binarySource && sourceType->isBlob())
                 {
                     if (type->isBlob())
@@ -1715,7 +1718,8 @@ private:
                             if (type->canTruncate())
                             {
                                 info.matchType = match_truncate;
-                                fixedDelta += sourceType->getMinSize()-type->getMinSize();
+                                if (((sourceFlags|destFlags) & RFTMinifblock) == 0)
+                                    fixedDelta += sourceType->getMinSize()-type->getMinSize();
                                 //DBGLOG("Increasing fixedDelta size by %d to %d for truncated field %d (%s)", sourceType->getMinSize()-type->getMinSize(), fixedDelta, idx, destRecInfo.queryName(idx));
                             }
                         }
@@ -1724,7 +1728,8 @@ private:
                             if (type->canExtend(info.fillChar))
                             {
                                 info.matchType = match_extend;
-                                fixedDelta += sourceType->getMinSize()-type->getMinSize();
+                                if (((sourceFlags|destFlags) & RFTMinifblock) == 0)
+                                    fixedDelta += sourceType->getMinSize()-type->getMinSize();
                                 //DBGLOG("Decreasing fixedDelta size by %d to %d for truncated field %d (%s)", type->getMinSize()-sourceType->getMinSize(), fixedDelta, idx, destRecInfo.queryName(idx));
                             }
                         }
@@ -1734,7 +1739,6 @@ private:
                     info.matchType = match_typecast;
                 if (deblob)
                     info.matchType |= match_deblob;
-                unsigned sourceFlags = sourceRecInfo.queryField(info.matchIdx)->flags;
                 if (sourceFlags & RFTMinifblock)
                     info.matchType |= match_inifblock;  // Avoids incorrect commoning up of adjacent matches
                 // MORE - could note the highest interesting fieldnumber in the source and not bother filling in offsets after that
@@ -1766,7 +1770,7 @@ private:
                     if (!destRecInfo.getFixedSize())
                     {
                         const RtlTypeInfo *type = field->type;
-                        if (type->isFixedSize())
+                        if (type->isFixedSize() && (field->flags & RFTMinifblock)==0)
                         {
                             //DBGLOG("Reducing estimated size by %d for (fixed size) omitted field %s", (int) type->getMinSize(), field->name);
                             fixedDelta += type->getMinSize();
@@ -1777,14 +1781,16 @@ private:
                     allUnmatched.append(idx);
                 }
             }
-            //DBGLOG("Source record contains %d bytes of omitted fixed size fields", fixedDelta);
+            //DBGLOG("Delta from fixed-size fields is %d bytes", fixedDelta);
         }
     }
     size32_t estimateNewSize(const RtlRow &sourceRow) const
     {
         //DBGLOG("Source record size is %d", (int) sourceRow.getRecordSize());
-        size32_t expectedSize = sourceRow.getRecordSize() - fixedDelta;
-        //DBGLOG("Source record size without omitted fixed size fields is %d", expectedSize);
+        size32_t expectedSize = sourceRow.getRecordSize();
+        assertex((int) expectedSize >= fixedDelta);
+        expectedSize -= fixedDelta;
+        //DBGLOG("Source record size without fixed delta is %d", expectedSize);
         ForEachItemIn(i, variableUnmatched)
         {
             unsigned fieldNo = variableUnmatched.item(i);
@@ -1814,8 +1820,12 @@ private:
                     // uft8 <-> string we could calculate here - but unlikely to be worth the effort.
                     // But it's fine for fixed size output fields, including truncate/extend
                     // We could also precalculate the expected delta if all omitted fields are fixed size - but not sure how likely/worthwhile that is.
-                    expectedSize += type->getMinSize() - sourceRow.getSize(matchField);
-                    //DBGLOG("Adjusting estimated size by (%d - %d) to %d for translated field %d (%s)", (int) sourceRow.getSize(matchField), type->getMinSize(), expectedSize, matchField, sourceRecInfo.queryName(matchField));
+                    auto minSize = type->getMinSize();
+                    auto sourceSize = sourceRow.getSize(matchField);
+                    expectedSize += minSize;
+                    assertex(expectedSize >= sourceSize);
+                    expectedSize -= sourceSize;
+                    //DBGLOG("Adjusting estimated size by (%d - %d) to %d for translated field %d (%s)", (int) sourceSize, minSize, expectedSize, matchField, sourceRecInfo.queryName(matchField));
                     break;
                 }
             }

system/security/LdapSecurity/ldapconnection.cpp

@@ -1798,6 +1798,21 @@ public:
                         DBGLOG("LDAP: User %s Must Reset Password", username);
                         user.setAuthenticateStatus(AS_PASSWORD_VALID_BUT_EXPIRED);
                     }
+                    else if (strstr(ldap_errstring, "data 533"))
+                    {
+                        DBGLOG("LDAP: User %s Account Disabled", username);
+                        user.setAuthenticateStatus(AS_ACCOUNT_DISABLED);
+                    }
+                    else if (strstr(ldap_errstring, "data 701"))
+                    {
+                        DBGLOG("LDAP: User %s Account Expired", username);
+                        user.setAuthenticateStatus(AS_ACCOUNT_EXPIRED);
+                    }
+                    else if (strstr(ldap_errstring, "data 775"))
+                    {
+                        DBGLOG("LDAP: User %s Account Locked Out", username);
+                        user.setAuthenticateStatus(AS_ACCOUNT_LOCKED);
+                    }
                     else
                     {
                         DBGLOG("LDAP: Authentication(1) (%c) for user %s failed - %s", isWorkunitDAToken(password) ? 't' :'f', username, ldap_err2string(rc));

system/security/shared/seclib.hpp

@@ -195,7 +195,10 @@ enum authStatus : int
     AS_UNEXPECTED_ERROR = 2,
     AS_INVALID_CREDENTIALS = 3,
     AS_PASSWORD_EXPIRED = 4,
-    AS_PASSWORD_VALID_BUT_EXPIRED = 5//user entered valid password, but authentication failed because it is expired
+    AS_PASSWORD_VALID_BUT_EXPIRED = 5,//user entered valid password, but authentication failed because it is expired
+    AS_ACCOUNT_DISABLED = 6,//valid username and password/credential are supplied but the account has been disabled
+    AS_ACCOUNT_EXPIRED = 7,//valid username and password/credential supplied but the account has expired
+    AS_ACCOUNT_LOCKED = 8,//valid username is supplied, but the account is locked out
 };
 
 class CDateTime;

testing/regress/ecl/diskgroupagg.ecl

@@ -0,0 +1,53 @@
+/*##############################################################################
+
+    HPCC SYSTEMS software Copyright (C) 2020 HPCC Systems®.
+
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+############################################################################## */
+
+//class=file
+
+import $.setup;
+prefix := setup.Files(false, false).QueryFilePrefix;
+
+rec := RECORD
+ unsigned4 id;
+ string s;
+END;
+
+// want all counts to be same, because output order of firstn, will
+// vary dependent on the configuration
+
+duplicates := 10;
+num := duplicates * 1000;
+
+dsname := prefix + 'gafile';
+indexname := prefix + 'gaindex';
+
+inds := DATASET(num, TRANSFORM(rec, SELF.id := COUNTER % duplicates; SELF.s := (string)COUNTER));
+ds := DATASET(dsname, rec, FLAT);
+i := INDEX(ds, {id}, {ds}, indexname);
+
+t1 := TABLE(ds, {id, c := COUNT(GROUP)}, id, FEW);
+t2 := TABLE(i, {id, c := COUNT(GROUP)}, id, FEW);
+
+SEQUENTIAL(
+ PARALLEL(
+  OUTPUT(inds, , dsname, OVERWRITE);
+  BUILD(i, OVERWRITE);
+ );
+ PARALLEL(
+  OUTPUT(CHOOSEN(TABLE(t1, {c}), 10));
+  OUTPUT(CHOOSEN(TABLE(t2, {c}), 10));
+ );
+);

testing/regress/ecl/ifblock-translate.ecl

@@ -0,0 +1,87 @@
+/*##############################################################################
+
+    HPCC SYSTEMS software Copyright (C) 2018 HPCC Systems®.
+
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+############################################################################## */
+
+// No point running on multiple engines - this is testing eclrtl
+
+//nothor
+//nohthor
+
+IMPORT std;
+
+STRING uc(STRING instr) := Std.Str.ToUpperCase(instr);
+
+rAA := RECORD
+    string002        SegmentID;
+    string012        FileSeqNum;
+  string50 aa1;
+END;
+
+rZZ := RECORD
+    string002        SegmentID;
+    string012        FileSeqNum;
+  string50 zz1;
+END;
+
+rAB := RECORD
+    string002        SegmentID;
+    string012        FileSeqNum;
+  string50 ab1;
+END;
+
+rAD := RECORD
+    string002        SegmentID;
+    string012        FileSeqNum;
+  string50 ad1;
+END;
+
+rIS := RECORD
+    string002        SegmentID;
+    string012        FileSeqNum;
+  string50 is1;
+END;
+
+rin := RECORD
+    string002        SegmentID;
+    string012        FileSeqNum;
+    IFBLOCK(uc(self.SegmentId) = 'AA')
+        rAA AND NOT [SegmentId, FileSeqNum] aa;
+    END;
+    IFBLOCK(uc(self.SegmentId) = 'ZZ')
+        rZZ AND NOT [SegmentId, FileSeqNum] zz;
+    END;
+    IFBLOCK(uc(self.SegmentId) = 'AB')
+        rAB AND NOT [SegmentId, FileSeqNum] ab;
+    END;
+    IFBLOCK(uc(self.SegmentId) = 'AD')
+        rAD AND NOT [SegmentId, FileSeqNum] ad;
+    END;
+    IFBLOCK(uc(self.SegmentId) = 'IS')
+        rIS AND NOT [SegmentId, FileSeqNum] is;
+    END;
+END;
+
+rout := RECORD
+    string        SegmentID;
+END;
+
+s := SERVICE
+   streamed dataset(rout) stransform(streamed dataset input) : eclrtl,pure,library='eclrtl',entrypoint='transformRecord',passParameterMeta(true);
+END;
+
+d := DATASET([{'AA', 'ABC', 'AAstring' }], rin);
+
+output(s.stransform(d));  

testing/regress/ecl/key/diskgroupagg.xml

@@ -0,0 +1,28 @@
+<Dataset name='Result 1'>
+</Dataset>
+<Dataset name='Result 2'>
+</Dataset>
+<Dataset name='Result 3'>
+ <Row><c>1000</c></Row>
+ <Row><c>1000</c></Row>
+ <Row><c>1000</c></Row>
+ <Row><c>1000</c></Row>
+ <Row><c>1000</c></Row>
+ <Row><c>1000</c></Row>
+ <Row><c>1000</c></Row>
+ <Row><c>1000</c></Row>
+ <Row><c>1000</c></Row>
+ <Row><c>1000</c></Row>
+</Dataset>
+<Dataset name='Result 4'>
+ <Row><c>1000</c></Row>
+ <Row><c>1000</c></Row>
+ <Row><c>1000</c></Row>
+ <Row><c>1000</c></Row>
+ <Row><c>1000</c></Row>
+ <Row><c>1000</c></Row>
+ <Row><c>1000</c></Row>
+ <Row><c>1000</c></Row>
+ <Row><c>1000</c></Row>
+ <Row><c>1000</c></Row>
+</Dataset>

testing/regress/ecl/key/ifblock-translate.xml

@@ -0,0 +1,3 @@
+<Dataset name='Result 1'>
+ <Row><segmentid>AA</segmentid></Row>
+</Dataset>

thorlcr/activities/diskread/thdiskreadslave.cpp

@@ -1164,10 +1164,19 @@ public:
     }
     virtual bool isGrouped() const override { return false; }
 // IRowStream
-    virtual void stop()
+    virtual void stop() override
     {
         if (partHandler)
             partHandler->stop();
+        if (aggregateStream)
+        {
+            aggregateStream->stop();
+            if (distributor)
+            {
+                distributor->disconnect(true);
+                distributor->join();
+            }            
+        }
         PARENT::stop();
     }
     CATCH_NEXTROW()

thorlcr/activities/indexread/thindexreadslave.cpp

@@ -1134,7 +1134,20 @@ public:
         done = true;
         return NULL;
     }
-    virtual void abort()
+    virtual void stop() override
+    {
+        if (aggregateStream)
+        {
+            aggregateStream->stop();
+            if (distributor)
+            {
+                distributor->disconnect(true);
+                distributor->join();
+            }            
+        }
+        PARENT::stop();
+    }
+    virtual void abort() override
     {
         CIndexReadSlaveBase::abort();
         if (merging)