Merge remote-tracking branch 'origin/candidate-9.14.x'

Signed-off-by: Gavin Halliday <[email protected]>

# Conflicts:
#	helm/hpcc/Chart.yaml
#	helm/hpcc/templates/_helpers.tpl
#	version.cmake

Author: ghalliday

common/thorhelper/thorsoapcall.cpp

@@ -2835,6 +2835,7 @@ class CWSCAsyncFor : implements IWSCAsyncFor, public CInterface, public CAsyncFo
                     break;
                 }
                 numRetries++;
+                master->logctx.noteStatistic(StNumSoapcallRetries, 1);
                 master->logctx.CTXLOG("Retrying: attempt %d of %d", numRetries, master->maxRetries);
                 master->activitySpanScope->recordException(e, false, false);
                 e->Release();

dali/base/dautils.cpp

@@ -3750,7 +3750,7 @@ class FileReadPropertiesUpdater : public CSimpleInterfaceOf<IFileReadPropertiesU
     }
 
 public:
-    FileReadPropertiesUpdater(IUserDescriptor * udesc) : udesc(udesc) {}
+    FileReadPropertiesUpdater(IUserDescriptor * _udesc) : udesc(_udesc) {}
 
     // Track and accumulate the readCost and numDiskReads to stats tracking map (to be written to properties later)
     // - if curReadCost is 0, it will be calculated using calcFileAccessCost

ecl/hthor/hthor.cpp

@@ -11042,10 +11042,6 @@ CHThorNewDiskReadBaseActivity::InputFileInfo * CHThorNewDiskReadBaseActivity::ex
         queryInheritSeparatorProp(*fileFormatOptions, "separator", options, "@csvSeparate");
         queryInheritProp(*fileFormatOptions, "terminator", options, "@csvTerminate");
         queryInheritProp(*fileFormatOptions, "escape", options, "@csvEscape");
-
-        //MORE: Remove before this is merged!
-        dbglogXML(fileFormatOptions);
-        dbglogXML(fileProviderOptions);
     }
 
     fileProviderOptions->setPropBool("@grouped", grouped);

roxie/ccd/ccdmain.cpp

@@ -885,6 +885,18 @@ int CCD_API roxie_main(int argc, const char *argv[], const char * defaultYaml)
         }
 #endif
 
+        //Add roxieMode configuration option, which can be used to set the defaults for various other options
+        bool isBatchRoxie = false;
+        const char * roxieMode = topology->queryProp("@roxieMode");
+        if (roxieMode && strieq(roxieMode, "batch"))
+            isBatchRoxie = true;
+
+        //Configure defaults based on whether the system is being used for batch or interactive queries
+        if (isBatchRoxie)
+        {
+            acknowledgeAllRequests = false;
+        }
+
         if (!topology->hasProp("@resolveLocally"))
             topology->setPropBool("@resolveLocally", !topology->hasProp("@daliServers"));
 
@@ -1361,6 +1373,15 @@ int CCD_API roxie_main(int argc, const char *argv[], const char * defaultYaml)
             setNodeFetchThresholdNs(topology->getPropInt64("@nodeFetchThresholdNs"));
         setIndexWarningThresholds(topology);
 
+        //Enabling the localNVMeCache will also by default enable remote file related optimizations
+        offset_t localNVMeCacheSize = topology->getPropInt64("@localNVMeCacheSize", 0);
+        bool usingRemoteStorage = (localNVMeCacheSize != 0);
+
+        unsigned inplaceSizeFactor = topology->getPropInt("@inplaceSizeFactor", usingRemoteStorage ? 0 : 100);
+        unsigned lz4SpeedFactor = topology->getPropInt("@lz4SpeedFactor", usingRemoteStorage ? 0 : 600);             // If lz4 is 5x faster then value should be 500
+        unsigned zStdSpeedFactor = topology->getPropInt("@zStdSpeedFactor", usingRemoteStorage ? 0 : 350);
+        setIndexScaling(inplaceSizeFactor, lz4SpeedFactor, zStdSpeedFactor);
+
         unsigned __int64 affinity = topology->getPropInt64("@affinity", 0);
         updateAffinity(affinity);
 

roxie/ccd/ccdserver.cpp

@@ -500,7 +500,7 @@ static const StatisticsMapping indexStatistics({StNumServerCacheHits, StNumIndex
 static const StatisticsMapping diskStatistics({StNumServerCacheHits, StNumDiskRowsRead, StNumDiskSeeks, StNumDiskAccepted,
                                                StNumDiskRejected, StSizeAgentReply, StTimeAgentWait, StTimeAgentQueue, StTimeAgentProcess, StTimeIBYTIDelay, StNumAckRetries, StNumAgentRequests, StSizeAgentRequests,
                                                StSizeContinuationData, StNumContinuationRequests }, actStatistics);
-static const StatisticsMapping soapStatistics({ StTimeSoapcall, StTimeSoapcallDNS, StTimeSoapcallConnect, StNumSoapcallConnectFailures }, actStatistics);
+static const StatisticsMapping soapStatistics({ StTimeSoapcall, StTimeSoapcallDNS, StTimeSoapcallConnect, StNumSoapcallConnectFailures, StNumSoapcallRetries }, actStatistics);
 static const StatisticsMapping groupStatistics({ StNumGroups, StNumGroupMax }, actStatistics);
 static const StatisticsMapping sortStatistics({ StTimeSortElapsed }, actStatistics);
 static const StatisticsMapping indexWriteStatistics({ StNumDuplicateKeys, StNumLeafCacheAdds, StNumNodeCacheAdds, StNumBlobCacheAdds }, actStatistics);
@@ -523,7 +523,7 @@ extern const StatisticsMapping accumulatedStatistics({StWhenFirstRow, StTimeLoca
                                                       StNumBloomAccepts, StNumBloomRejects, StNumBloomSkips,
                                                       StNumNodeDiskFetches, StNumLeafDiskFetches, StNumBlobDiskFetches,
                                                       StNumDiskRejected, StSizeAgentReply, StTimeAgentWait,
-                                                      StTimeSoapcall, StTimeSoapcallDNS, StTimeSoapcallConnect, StNumSoapcallConnectFailures,
+                                                      StTimeSoapcall, StTimeSoapcallDNS, StTimeSoapcallConnect, StNumSoapcallConnectFailures, StNumSoapcallRetries,
                                                       StNumGroups,
                                                       StTimeSortElapsed,
                                                       StNumDuplicateKeys,

system/jhtree/jhinplace.cpp

@@ -39,6 +39,36 @@
 //#define TRACE_BUILDING_STATS
 #endif
 
+// A setting that can be scaled from 0 to 100 to track how the extra in memory saving from inplace indexes affects the performance
+// of roxie.  Memory allocated = newSize + (oldSize - newSize) * inplaceSizeFactor/100;
+static unsigned inplaceSizeFactor = 0;
+
+// Some settings that can be used to scale the time taken to expand nodes using the new inplace indexes.
+//
+// EstimatedTime = (TimeTakenToExpand / newPayloadSize) * oldSize * XXXSpeedFactor /100
+//
+// i.e. Estimate the time it would have taken to expand the original payload, assuming lzw compression
+// takes speedFactor/100 times longer than the new compression.
+//
+// This should also be used in the upcoming hybrid compression (e.g. hybrid:zstd) which uses the key compression for branches,
+// but the legacy compression format for leaves.
+//
+// A factor < 100 may still add time if there is a noticeable reduction in the payload data size.
+
+static unsigned lz4SpeedFactor = 0;
+static unsigned zStdSpeedFactor = 0;
+static bool adjustExpansionTime = false;
+
+void setIndexScaling(unsigned _inplaceSizeFactor, unsigned _lz4SpeedFactor, unsigned _zStdSpeedFactor)
+{
+    inplaceSizeFactor = _inplaceSizeFactor;
+    lz4SpeedFactor = _lz4SpeedFactor;
+    zStdSpeedFactor = _zStdSpeedFactor;
+    adjustExpansionTime = (lz4SpeedFactor != 0) || (zStdSpeedFactor != 0);
+}
+
+//---------------------------------------------------------------------------------------------------------------------
+
 static constexpr size32_t minRepeatCount = 2;       // minimum number of times a 0x00 or 0x20 is repeated to generate a special opcode
 static constexpr size32_t minRepeatXCount = 3;      // minimum number of times a byte is repeated to generate a special opcode
 static constexpr size32_t maxQuotedCount = 31 + 256; // maximum number of characters that can be quoted in a row
@@ -1665,6 +1695,8 @@ int CJHInplaceTreeNode::locateGT(const char * search, unsigned minIndex) const
 }
 
 
+constexpr static bool traceInplaceLoadStats = false;
+
 void CJHInplaceTreeNode::load(CKeyHdr *_keyHdr, const void *rawData, offset_t _fpos, bool needCopy)
 {
     CJHSearchNode::load(_keyHdr, rawData, _fpos, needCopy);
@@ -1675,20 +1707,29 @@ void CJHInplaceTreeNode::load(CKeyHdr *_keyHdr, const void *rawData, offset_t _f
 
     const byte * nullRow = nullptr; //MORE: This should be implemented
     unsigned numKeys = hdr.numKeys;
+    unsigned originalKeyedSize = keyCompareLen * numKeys;
     if (numKeys)
     {
+         // only time the follow code if we are going to try and match the old timing.
+        CCycleTimer expansionTimer(traceInplaceLoadStats || (adjustExpansionTime && isLeaf()));
+
         size32_t len = hdr.keyBytes;
         size32_t copyLen = len;
         const byte * originalData = ((const byte *)rawData) + sizeof(hdr);
         const byte * originalPayload = nullptr;
         bool keepCompressedPayload = true;
+        size32_t actualKeyedSize = len;
+
+        //For branches payloadCompression will always be COMPRESS_METHOD_NONE
+        CompressionMethod payloadCompression = COMPRESS_METHOD_NONE;
         if (isLeaf())
         {
             //Always calculate payload location so we can perform a consistency check later on.
             originalPayload = queryPayload(originalData);
             if (originalPayload)
             {
-                CompressionMethod payloadCompression = (CompressionMethod)*originalPayload;
+                actualKeyedSize = originalPayload - originalData;
+                payloadCompression = (CompressionMethod)*originalPayload;
                 switch (payloadCompression)
                 {
                 case COMPRESS_METHOD_NONE:
@@ -1699,19 +1740,27 @@ void CJHInplaceTreeNode::load(CKeyHdr *_keyHdr, const void *rawData, offset_t _f
                     expandPayloadOnDemand = dynamicPayloadExpansion;
                     keepCompressedPayload = expandPayloadOnDemand;
                     if (!expandPayloadOnDemand)
-                        copyLen = (originalPayload - originalData);
+                        copyLen = actualKeyedSize;
                     break;
                 }
             }
             else
+            {
+                //if no payload then actualKeySize is already set....
                 expandPayloadOnDemand = false;
+            }
         }
 
-        const size32_t padding = 8 - 1; // Ensure that unsigned8 values can be read "legally"
+        size32_t padding = 8 - 1; // Ensure that unsigned8 values can be read "legally"
+
+        //Allow the memory used by the inplace indexes to be adjusted to explore the trend of the benefits
+        if (inplaceSizeFactor && (actualKeyedSize < originalKeyedSize))
+            padding += (originalKeyedSize - actualKeyedSize) * inplaceSizeFactor / 100;
+
         keyBuf = (char *) allocMem(copyLen + padding);
         memcpy(keyBuf, originalData, copyLen);
         memset(keyBuf+copyLen, 0, padding);
-        expandedSize = copyLen;
+        expandedSize = copyLen+padding;
 
         /**** If any of the following code changes queryPayload() must also be changed. ******/
 
@@ -1814,6 +1863,53 @@ void CJHInplaceTreeNode::load(CKeyHdr *_keyHdr, const void *rawData, offset_t _f
             assertex((data - (const byte *)keyBuf) == len);
         else
             assertex((data - originalData) == len);
+
+        //Branch nodes do not use LZW compression - so this will not attempt to adjust them.
+        if ((adjustExpansionTime || traceInplaceLoadStats) && (payloadCompression != COMPRESS_METHOD_NONE) && !expandPayloadOnDemand)
+        {
+            __uint64 timeTakenNs = expansionTimer.elapsedNs();
+            unsigned scaling = 0;
+            switch (payloadCompression)
+            {
+            case COMPRESS_METHOD_LZW:
+            case COMPRESS_METHOD_LZW_LITTLE_ENDIAN:
+                scaling = 100; // only adjustment will be due to not expanding the keyed portion.
+                break;
+            case COMPRESS_METHOD_LZ4:
+            case COMPRESS_METHOD_LZ4HC:
+            case COMPRESS_METHOD_LZ4S:
+            case COMPRESS_METHOD_LZ4SHC:
+            case COMPRESS_METHOD_LZ4HC3:
+                scaling = lz4SpeedFactor;
+                break;
+            case COMPRESS_METHOD_ZSTDS:
+                scaling = zStdSpeedFactor;
+                break;
+            }
+
+            if (scaling)
+            {
+                size32_t actualSizeExpanded = expandedSize - padding - actualKeyedSize;     // How much data was expanded in the payload
+                size32_t originalSizeExpanded = actualSizeExpanded + originalKeyedSize;     // How much data would there have been if the whole node was compressed?
+
+                if (actualSizeExpanded)
+                {
+                    //This scaling will not work sensibly if keys are not expanded on demand.
+                    // expected = (TimeTakenToExpand / newPayloadSize) * oldSize * XXXSpeedFactor /100;
+                    // rearrange so the divisions happen last
+                    __uint64 expectedTime = (timeTakenNs * originalSizeExpanded * scaling) / actualSizeExpanded / 100;
+
+                    //Sanity check to avoid pathological times caused by context switches etc.
+                    if (expectedTime > 500'000)
+                        expectedTime = 500'000;
+
+                    if (traceInplaceLoadStats)
+                        DBGLOG("InplaceLoad: %s originalSize(%u) actualSize(%u), expectedTime(%llu), actualTime(%llu)", translateFromCompMethod(payloadCompression), originalSizeExpanded, actualSizeExpanded, expectedTime, timeTakenNs);
+                    else if (expectedTime > timeTakenNs)
+                        NanoSleep(expectedTime- timeTakenNs);
+                }
+            }
+        }
     }
 }
 

system/jhtree/jhtree.hpp

@@ -299,6 +299,8 @@ extern jhtree_decl bool isIndexFile(IFile *filename);
 
 extern jhtree_decl IIndexLookup *createIndexLookup(IKeyManager *keyManager);
 
+extern jhtree_decl void setIndexScaling(unsigned _inplaceSizeFactor, unsigned _lz4SpeedFactor, unsigned _zStdSpeedFactor);
+
 #define JHTREE_KEY_NOT_SORTED JHTREE_ERROR_START
 #define JHTREE_KEY_UNKNOWN_COMPRESSION (JHTREE_ERROR_START+1)
 

system/jlib/jstatcodes.h

@@ -360,6 +360,7 @@ enum StatisticKind
     StTimeQueryConsume,
     StCycleQueryConsumeCycles,
     StNumSuccesses,
+    StNumSoapcallRetries,
     StMax,
 
     //For any quantity there is potentially the following variants.

system/jlib/jstats.cpp

@@ -1011,10 +1011,10 @@ static const constexpr StatisticMeta statsMetaData[StMax] = {
     { NUMSTAT(BloomRejects), "The number of times a bloom filter rejects an index lookup" },
     { NUMSTAT(BloomSkips), "The number of times a bloom filter cannot filter an index lookup" },
     { NUMSTAT(Accepts), "The number of items accepted for processing" },
-    { NUMSTAT(Waits), "The number of times a component waits for a entry on a queue" },
+    { NUMSTAT(Waits), "The number of times a component waits for an entry on a queue" },
     { TIMESTAT(Provision), "The total time spent provisioning a component" },
     { CYCLESTAT(Provision) },
-    { COSTSTAT(Start), "The cost assocaiated with starting a component or operation" },
+    { COSTSTAT(Start), "The cost associated with starting a component or operation" },
     { TIMESTAT(WaitSuccess), "The time waiting for an item on a queue, when an item was eventually received" },
     { CYCLESTAT(WaitSuccess) },
     { TIMESTAT(WaitFailure), "The time waiting for an item on a queue, when no item was received" },
@@ -1027,6 +1027,7 @@ static const constexpr StatisticMeta statsMetaData[StMax] = {
     { TIMESTAT(QueryConsume), "The total time spent consuming and processing a query input" },
     { CYCLESTAT(QueryConsume) },
     { NUMSTAT(Successes), "The number of times something was successful" },
+    { NUMSTAT(SoapcallRetries), "The number of times a soapcall request retries" },
 };
 
 static MapStringTo<StatisticKind, StatisticKind> statisticNameMap(true);

thorlcr/thorutil/thormisc.cpp

@@ -80,7 +80,7 @@ static Owned<IMPtagAllocator> ClusterMPAllocator;
 // stat. mappings shared between master and slave activities
 const StatisticsMapping spillStatistics({StTimeSpillElapsed, StTimeSortElapsed, StNumSpills, StSizeSpillFile, StSizePeakTempDisk});
 const StatisticsMapping executeStatistics({StWhenFirstRow, StTimeElapsed, StTimeTotalExecute, StTimeLocalExecute, StTimeBlocked});
-const StatisticsMapping soapcallStatistics({StTimeSoapcall, StTimeSoapcallDNS, StTimeSoapcallConnect, StNumSoapcallConnectFailures});
+const StatisticsMapping soapcallStatistics({StTimeSoapcall, StTimeSoapcallDNS, StTimeSoapcallConnect, StNumSoapcallConnectFailures, StNumSoapcallRetries});
 const StatisticsMapping basicActivityStatistics({StNumParallelExecute, StTimeLookAhead}, executeStatistics, spillStatistics);
 const StatisticsMapping groupActivityStatistics({StNumGroups, StNumGroupMax}, basicActivityStatistics);
 const StatisticsMapping indexReadFileStatistics({}, diskReadRemoteStatistics, jhtreeCacheStatistics);