Merge branch 'candidate-7.4.x' into candidate-7.6.x

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

ecl/eclagent/eclagent.cpp

@@ -1121,7 +1121,7 @@ void EclAgent::setResultData(const char * stepname, unsigned sequence, int len,
 
 void EclAgent::doSetResultString(type_t type, const char *name, unsigned sequence, int len, const char *val)
 {
-    LOG(MCsetresult, unknownJob, "setResultString(%s,%d,'%.*s')", nullText(name), sequence, len, val);
+    LOG(MCsetresult, unknownJob, "setResultString(%s,%d,(%d bytes))", nullText(name), sequence, len);
     WorkunitUpdate w = updateWorkUnit();
     Owned<IWUResult> r = updateWorkUnitResult(w, name, sequence);
     if (r)

ecl/hqlcpp/hqlhtcpp.cpp

@@ -11042,7 +11042,7 @@ ABoundActivity * HqlCppTranslator::doBuildActivityOutput(BuildCtx & ctx, IHqlExp
             OwnedHqlExpr serializedRecord = getSerializedForm(dataset->queryRecord(), diskAtom);
             buildMetaMember(instance->classctx, serializedRecord, grouped && !ignoreGrouped, "queryDiskRecordSize");
         }
-        buildClusterHelper(instance->classctx, expr);
+        buildClusterHelper(instance->startctx, expr);
 
         //Both csv write and pipe with csv/xml format
         if (csvAttr)

plugins/logging/logging.cpp

@@ -64,7 +64,14 @@ LOGGING_API bool getECLPluginDefinition(ECLPluginDefinitionBlock *pb)
 
 LOGGING_API void LOGGING_CALL logDbgLog(unsigned srcLen, const char * src)
 {
-    DBGLOG("%.*s", srcLen, src);
+    StringBuffer log(srcLen, src);
+    StringArray loglines;
+    log.replace('\r', ' ');
+    loglines.appendList(log, "\n", false);
+    ForEachItemIn(idx, loglines)
+    {
+        DBGLOG("%s", loglines.item(idx));
+    }
 }
 
 LOGGING_API char *  LOGGING_CALL logGetGlobalId(ICodeContext *ctx)

thorlcr/activities/keyedjoin/thkeyedjoin.cpp

@@ -24,6 +24,7 @@
 
 #include "../fetch/thfetchcommon.hpp"
 #include "../hashdistrib/thhashdistrib.ipp"
+#include "thkeyedjoincommon.hpp"
 #include "thkeyedjoin.ipp"
 #include "jhtree.hpp"
 
@@ -46,7 +47,6 @@ class CKeyedJoinMaster : public CMasterActivity
     // CMap contains mappings and lists of parts for each slave
     class CMap
     {
-        static const unsigned partMask = 0x00ffffff;
     public:
         std::vector<unsigned> allParts;
         std::vector<std::vector<unsigned>> slavePartMap; // vector of slave parts (IPartDescriptor's slavePartMap[<slave>] serialized to each slave)
@@ -152,7 +152,7 @@ class CKeyedJoinMaster : public CMasterActivity
                         for (unsigned c=0; c<part->numCopies(); c++)
                         {
                             INode *partNode = part->queryNode(c);
-                            unsigned partCopy = p | (c << 24);
+                            unsigned partCopy = p | (c << partBits);
                             unsigned start=nextGroupStartPos;
                             unsigned gn=start;
                             do
@@ -180,6 +180,14 @@ class CKeyedJoinMaster : public CMasterActivity
                                             nextGroupStartPos = gn+1;
                                             if (nextGroupStartPos == groupSize)
                                                 nextGroupStartPos = 0;
+
+                                            /* NB: normally if the part is within the cluster, the copy will be 0 (i.e. primary)
+                                             * But it's possible that a non-primary copy from another logical cluster is local to
+                                             * this cluster, in which case, must capture which copy it is here in the map, so the
+                                             * slaves can send the requests to the correct slave and tell it to deal with the
+                                             * correct copy.
+                                             */
+                                            mappedPos |= (c << slaveBits); // encode which copy into mappedPos
                                         }
                                         else if (allLocal) // all slaves get all locally accessible parts
                                             slaveParts.push_back(partCopy);

thorlcr/activities/keyedjoin/thkeyedjoincommon.hpp

@@ -50,4 +50,11 @@ enum KJServiceCmds:byte { kjs_nop, kjs_keyopen, kjs_keyread, kjs_keyclose, kjs_f
 enum KJFetchFlags:byte { kjf_nop=0x0, kjf_compressed=0x1, kjf_encrypted=0x2 };
 enum KJServiceErrorCode:byte { kjse_nop, kjse_exception, kjse_unknownhandle };
 
+constexpr unsigned partBits = 24;
+constexpr unsigned partMask = 0x00ffffff;
+// the same as part, but for clarify has own symbols
+constexpr unsigned slaveBits = 24;
+constexpr unsigned slaveMask = 0x00ffffff;
+
+
 #endif

thorlcr/activities/keyedjoin/thkeyedjoinslave.cpp

@@ -1008,7 +1008,7 @@ class CKeyedJoinSlave : public CSlaveActivity, implements IJoinProcessor, implem
             for (auto &partCopy : parts)
             {
                 unsigned partNo = partCopy & partMask;
-                unsigned copy = partCopy >> 24;
+                unsigned copy = partCopy >> partBits;
                 IPartDescriptor &pd = allParts->item(partNo);
                 RemoteFilename rfn;
                 pd.getFilename(copy, rfn);
@@ -1215,7 +1215,6 @@ class CKeyedJoinSlave : public CSlaveActivity, implements IJoinProcessor, implem
                 limiter->dec(); // unblocks any requests to start lookup threads
         }
     };
-    static const unsigned partMask = 0x00ffffff;
 
     class CKeyLookupLocalBase : public CLookupHandler
     {
@@ -1336,7 +1335,7 @@ class CKeyedJoinSlave : public CSlaveActivity, implements IJoinProcessor, implem
         {
             unsigned partCopy = parts[selected];
             unsigned partNo = partCopy & partMask;
-            unsigned copy = partCopy >> 24;
+            unsigned copy = partCopy >> partBits;
             IKeyManager *&keyManager = keyManagers[selected];
             if (!keyManager) // delayed until actually needed
             {
@@ -1366,7 +1365,7 @@ class CKeyedJoinSlave : public CSlaveActivity, implements IJoinProcessor, implem
                 for (auto &partCopy: parts)
                 {
                     unsigned partNo = partCopy & partMask;
-                    unsigned copy = partCopy >> 24;
+                    unsigned copy = partCopy >> partBits;
                     Owned<IKeyIndex> keyIndex = activity.createPartKeyIndex(partNo, copy, false);
                     partKeySet->addIndex(keyIndex.getClear());
                 }
@@ -1546,7 +1545,7 @@ class CKeyedJoinSlave : public CSlaveActivity, implements IJoinProcessor, implem
         {
             unsigned partCopy = parts[selected];
             unsigned partNo = partCopy & partMask;
-            unsigned copy = partCopy >> 24;
+            unsigned copy = partCopy >> partBits;
 
             unsigned numRows = processing.ordinality();
 
@@ -1701,7 +1700,7 @@ class CKeyedJoinSlave : public CSlaveActivity, implements IJoinProcessor, implem
         {
             unsigned partCopy = parts[selected];
             unsigned partNo = partCopy & partMask;
-            unsigned copy = partCopy >> 24;
+            unsigned copy = partCopy >> partBits;
 
             ScopedAtomic<unsigned __int64> diskRejected(activity.statsArr[AS_DiskRejected]);
             ScopedAtomic<unsigned __int64> diskSeeks(activity.statsArr[AS_DiskSeeks]);
@@ -1852,7 +1851,7 @@ class CKeyedJoinSlave : public CSlaveActivity, implements IJoinProcessor, implem
         {
             unsigned partCopy = parts[selected];
             unsigned partNo = partCopy & partMask;
-            unsigned copy = partCopy >> 24;
+            unsigned copy = partCopy >> partBits;
 
             CMessageBuffer msg;
             prepAndSend(msg, processing, selected, partNo, copy);
@@ -2140,6 +2139,7 @@ class CKeyedJoinSlave : public CSlaveActivity, implements IJoinProcessor, implem
 
     CriticalSection fetchFileCrit;
     std::vector<PartIO> openFetchParts;
+    const unsigned unknownCopyNum = 0xff; // in a partCopy denotes that a copy is unknown.
 
     PartIO getFetchPartIO(unsigned partNo, unsigned copy, bool compressed, bool encrypted)
     {
@@ -2612,26 +2612,37 @@ class CKeyedJoinSlave : public CSlaveActivity, implements IJoinProcessor, implem
         }
         return nullptr;
     }
-
     bool transferToDoneList(CJoinGroup *joinGroup, bool markBlocked) // NB: always coming from pendingJoinGroupList
     {
         doneJoinGroupList.addToTail(joinGroup);
         pendingKeyLookupLimiter.dec();
         return doneListLimiter.preIncNonBlocking(markBlocked);
     }
-
     void addPartToHandler(CHandlerContainer &handlerContainer, const std::vector<unsigned> &partToSlaveMap, unsigned partCopy, HandlerType hType, std::vector<unsigned> &handlerCounts, std::vector<std::vector<CLookupHandler *>> &slaveHandlers, std::vector<unsigned> &slaveHandlersRR)
     {
         // NB: This is called in partNo ascending order
 
         unsigned partNo = partCopy & partMask;
-        unsigned copy = partCopy >> 24;
+        unsigned copy = partCopy >> partBits;
         unsigned slave = 0;
         if (partToSlaveMap.size())
         {
             slave = partToSlaveMap[partNo];
             if (NotFound == slave) // part not local to cluster, part is handled locally/directly.
                 slave = handlerCounts.size()-1; // last one reserved for out of cluster part handling.
+            else
+            {
+                if (unknownCopyNum == copy)// this means that this is a part for a remote slave, and copy is unknown
+                {
+                    /* The partToSlaveMap encodes the target slave and the copy.
+                     * Extract the copy, so that the correct copy is used when request
+                     * arrives at the remote slave.
+                     */
+                    copy = slave >> slaveBits;
+                    partCopy = partNo | (copy << partBits);
+                }
+                slave = slave & slaveMask;
+            }
         }
         unsigned max = queryMaxHandlers(hType);
         unsigned &handlerCount = handlerCounts[slave];
@@ -2723,11 +2734,11 @@ class CKeyedJoinSlave : public CSlaveActivity, implements IJoinProcessor, implem
                     partNo = partCopy & partMask;
                     if (superFdesc)
                     {
-                        unsigned copy = partCopy >> 24;
+                        unsigned copy = partCopy >> partBits;
                         unsigned subfile, subpartnum;
                         superFdesc->mapSubPart(partNo, subfile, subpartnum);
                         partNo = superWidth*subfile+subpartnum;
-                        partCopy = partNo | (copy << 24);
+                        partCopy = partNo | (copy << partBits);
                     }
                 }
                 else
@@ -2740,7 +2751,10 @@ class CKeyedJoinSlave : public CSlaveActivity, implements IJoinProcessor, implem
                 while (p<partNo)
                 {
                     if (remoteLookup) // NB: only relevant if ,LOCAL and only some parts avail. otherwise if !remoteLookup all parts will have been sent
-                        addPartToHandler(handlerContainer, partToSlaveMap, p, missingHandlerType, handlerCounts, slaveHandlers, slaveHandlersRR);
+                    {
+                        unsigned remotePartCopy = p | (unknownCopyNum << partBits); // copy will be looked up via map in addPartToHandler
+                        addPartToHandler(handlerContainer, partToSlaveMap, remotePartCopy, missingHandlerType, handlerCounts, slaveHandlers, slaveHandlersRR);
+                    }
                     else // no handler if local KJ and part not local
                         handlerContainer.partIdxToHandler.push_back(nullptr);
                     ++p;

tools/hidl/hidlcomp.cpp

@@ -5743,6 +5743,13 @@ void EspServInfo::write_esp_binding()
         StrBuffer minVer;
         bool hasMinVer = mthi->getMetaVerInfo("min_ver", minVer);
 
+        if (hasMinVer) {
+            outs(2, "if (!clientVer) {\n");
+            outf(3, "clientVer = %s;\n", minVer.str());
+            outs(3, "ctx->setClientVersion(clientVer);\n");
+            outs(2, "}\n");
+        }
+
         bool bHandleExceptions = 0 != mthi->getMetaInt("exceptions_inline", 0);
         if (!bHandleExceptions)
             bHandleExceptions = 0 != getMetaInt("exceptions_inline", 0);
@@ -5799,18 +5806,20 @@ void EspServInfo::write_esp_binding()
         {
             if (hasMinVer)
             {
-                outf("\t\tif (clientVer!=-1.0 && clientVer<%s)\n", minVer.str());
-                outs("\t\t\tthrow MakeStringException(-1, \"Client version is too old (can't pass exception to client)\");\n");
+                outf(2, "if (clientVer!=-1.0 && clientVer<%s)\n", minVer.str());
+                outf(3, "throw MakeStringException(-1, \"This method is not supported in version %%g, minimum version is %s. Please update your client application.\", clientVer);\n", minVer.str());
             }
             if (mthi->getMetaInt("do_not_log",0))
-                outs("\t\tcontext.queryRequestParameters()->setProp(\"do_not_log\",1);\n");
+                outs(2, "context.queryRequestParameters()->setProp(\"do_not_log\",1);\n");
 
-            if (servicefeatureurl.length() != 0)
-                outf("\t\tif( accessmap.ordinality() > 0 )\n\t\t\tonFeaturesAuthorize(context, accessmap, \"%s\", \"%s\");\n", name_, mthi->getName());
-            outf("\t\tiserv->on%s(*rpc_call->queryContext(), *esp_request, *esp_response);\n", mthi->getName());
+            if (servicefeatureurl.length() != 0) {
+                outs(2, "if( accessmap.ordinality() > 0 )\n");
+                outf(3, "onFeaturesAuthorize(context, accessmap, \"%s\", \"%s\");\n", name_, mthi->getName());
+            }
+            outf(2, "iserv->on%s(*rpc_call->queryContext(), *esp_request, *esp_response);\n", mthi->getName());
             if (clearCacheGroupIDs.length() > 0)
-                outf("\t\tclearCacheByGroupID(\"%s\");\n", clearCacheGroupIDs.str());
-            outs("\t\tresponse->set_status(SOAP_OK);\n");
+                outf(2, "clearCacheByGroupID(\"%s\");\n", clearCacheGroupIDs.str());
+            outs(2, "response->set_status(SOAP_OK);\n");
         }
 
         outf("\t\tresponse->set_name(\"%s\");\n", mthi->getResp());