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

Signed-off-by: Gordon Smith <GordonJSmith@gmail.com>

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

Author: GordonSmith

common/eventconsumption/README.md

@@ -320,9 +320,13 @@ If algorithm estimation is enabled, the value is replaced. If the node will not
 
 
 
-### CMetaInfoState
+### CMetaInfoState::CCollector
 
-A concrete implementation of `IEventVisitationLink` that observes all `FileInformation` and `QueryStart` events. The relationships between `FileId` and `Path` attributes, and also `EventTraceId` and `ServiceName` attributes, are retained and made available to operations and other event visitors. When visited before a filter, the state information is available to all operations and visitors even if the filter suppresses the observed events.
+`CMetaInfoState` provides a runtime repository of event-defined state information, including values from the `FileInformation` and `QueryStart` events. As a concrete implementation of `IEventVisitationLink`, `CCollector` is a companion class that extracts event data for the repository.
+
+To avoid data loss, this link must precede all other links. It is the responsibility of event iterating code to ensure this requirement is satisfied.
+
+The repository is available to all operations and all other visitors. The collector itself is not.
 
 # CEventVisitationLinkTester
 

common/eventconsumption/eventindexhotspot.cpp

@@ -138,13 +138,12 @@ class CHotspotEventVisitor : public CInterfaceOf<IEventVisitor>
         if (event.hasAttribute(EvAttrFileId))
         {
             EventType type = event.queryType();
-            if (type != MetaFileInformation && observedEvents.count(type) == 0)
+            if (type == MetaFileInformation || observedEvents.count(type) == 0)
                 return true;
             __uint64 fileId = event.queryNumericValue(EvAttrFileId);
             auto [it, inserted] = activity.try_emplace(fileId, *this, fileId);
             NodeKind nodeKind = queryIndexNodeKind(event);
-            if (observedEvents.count(type))
-                it->second.recordEvent(event.queryNumericValue(EvAttrFileOffset), nodeKind);
+            it->second.recordEvent(event.queryNumericValue(EvAttrFileOffset), nodeKind);
         }
         return true;
     }

common/eventconsumption/eventindexhotspotbuckets.cpp

@@ -148,7 +148,12 @@ class CAllBucketVisitor : public CStreamingBucketVisitor
             lines.append("  ").append(mapNodeKind(nodeKind)).append('\n');
             firstOfKind[nodeKind] = false;
         }
-        lines.append("  - first-node: ").append(bucket2page(bucket, currentGranularity)).append('\n');
+
+        auto firstNode = bucket2page(bucket, currentGranularity);
+        auto lastNode = bucket2page(bucket + 1, currentGranularity) - 1;
+
+        lines.append("  - node-range: ").append(firstNode).append('-').append(lastNode).append('\n');
+        lines.append("    offset-range: ").append(page2Offset(firstNode, defaultPageBits)).append('-').append(page2Offset(lastNode + 1, defaultPageBits) - 1).append('\n');
         lines.append("    events: ").append(stat).append('\n');
         out->put(lines.length(), lines.str());
     }
@@ -281,7 +286,11 @@ class CTopBucketVisitor : public CStreamingBucketVisitor
         lines.append("    bucket:\n");
         for (const Hit& hit : info.hits)
         {
-            lines.append("    - first-node: ").append(bucket2page(hit.second, currentGranularity)).append('\n');
+            auto firstNode = bucket2page(hit.second, currentGranularity);
+            auto lastNode = bucket2page(hit.second + 1, currentGranularity) - 1;
+
+            lines.append("    - node-range: ").append(firstNode).append('-').append(lastNode).append('\n');
+            lines.append("      offset-range: ").append(page2Offset(firstNode, defaultPageBits)).append('-').append(page2Offset(lastNode + 1, defaultPageBits) - 1).append('\n');
             lines.append("      events: ").append(hit.first).append('\n');
         }
         if (info.droppedCount)

common/eventconsumption/eventmetaparser.cpp

@@ -18,11 +18,11 @@
 #include "eventmetaparser.hpp"
 #include "jevent.hpp"
 
-void CMetaInfoState::configure(const IPropertyTree& config)
+void CMetaInfoState::CCollector::configure(const IPropertyTree& config)
 {
 }
 
-bool CMetaInfoState::visitEvent(CEvent& event)
+bool CMetaInfoState::CCollector::visitEvent(CEvent& event)
 {
     switch (event.queryType())
     {
@@ -32,8 +32,8 @@ bool CMetaInfoState::visitEvent(CEvent& event)
             const char* path = event.queryTextValue(EvAttrPath);
             if (!isEmptyString(path))
             {
-                fileIdToPath[fileId].set(path);
-                pathToFileId[path] = fileId;
+                metaState->fileIdToPath[fileId].set(path);
+                metaState->pathToFileId[path] = fileId;
             }
         }
         break;
@@ -44,7 +44,7 @@ bool CMetaInfoState::visitEvent(CEvent& event)
                 const char* traceId = event.queryTextValue(EvAttrEventTraceId);
                 const char* serviceName = event.queryTextValue(EvAttrServiceName);
                 if (!isEmptyString(traceId) && !isEmptyString(serviceName))
-                    traceIdToService.emplace(traceId, serviceName);
+                    metaState->traceIdToService.emplace(traceId, serviceName);
             }
         }
         break;
@@ -56,6 +56,16 @@ bool CMetaInfoState::visitEvent(CEvent& event)
     return true;
 }
 
+CMetaInfoState::CCollector::CCollector(CMetaInfoState& _metaState)
+    : metaState(&_metaState)
+{
+}
+
+IEventVisitationLink* CMetaInfoState::getCollector()
+{
+    return new CCollector(*this);
+}
+
 const char* CMetaInfoState::queryFilePath(__uint64 fileId) const
 {
     auto it = fileIdToPath.find(fileId);

common/eventconsumption/eventmetaparser.hpp

@@ -24,17 +24,23 @@
 #include <string>
 
 // Visitor that parses and caches file ID to path mappings and trace ID to service name mappings
-class event_decl CMetaInfoState : public CInterfaceOf<IEventVisitationLink>
+class event_decl CMetaInfoState : public CInterface
 {
-public:
-    CMetaInfoState() = default;
-    ~CMetaInfoState() = default;
-
-    // IEventVisitationLink interface
-    IMPLEMENT_IEVENTVISITATIONLINK
-    virtual void configure(const IPropertyTree& config) override;
-    virtual bool visitEvent(CEvent& event) override;
+    class event_decl CCollector : public CInterfaceOf<IEventVisitationLink>
+    {
+    public: // IEventVisitor
+        virtual bool visitEvent(CEvent& event) override;
+    public: // IEventVisitationLink
+        IMPLEMENT_IEVENTVISITATIONLINK
+        virtual void configure(const IPropertyTree& config) override;
+    public:
+        CCollector(CMetaInfoState& _metaState);
+    private:
+        Linked<CMetaInfoState> metaState;
+    };
 
+public:
+    IEventVisitationLink* getCollector();
     // Accessor functions for file ID to path mappings
     const char* queryFilePath(__uint64 fileId) const;
     bool hasFileMapping(__uint64 fileId) const;
@@ -58,4 +64,4 @@ class event_decl CMetaInfoState : public CInterfaceOf<IEventVisitationLink>
 
     // Trace ID to service name mappings (from EventQueryStart events)
     std::unordered_map<std::string, std::string> traceIdToService;
-};
+};

common/eventconsumption/eventoperation.cpp

@@ -77,10 +77,8 @@ bool CEventConsumingOp::traverseEvents(const char* path, IEventVisitor& visitor)
         model->setNextLink(*actual);
         actual = model;
     }
+    Owned<IEventVisitationLink> metaCollector = metaState->getCollector();
+    metaCollector->setNextLink(*actual);
 
-    // Always include the meta information parser as the first link in the chain
-    // to ensure MetaFileInformation and EventQueryStart events are captured
-    metaState->setNextLink(*actual);
-
-    return readEvents(path, *metaState);
+    return readEvents(path, *metaCollector);
 }

common/eventconsumption/eventoperation.h

@@ -47,6 +47,7 @@ class event_decl CEventConsumingOp : public CInterface
     Owned<CMetaInfoState> metaState;
     StringAttr inputPath;
     Linked<IBufferedSerialOutputStream> out;
+private:
     Owned<IEventFilter> filter;
     Owned<IEventModel> model;
 };

common/eventconsumption/eventunittests.cpp

@@ -186,8 +186,9 @@ void testEventVisitationLinks(const IPropertyTree& inputTree, const IPropertyTre
             }
 
             // Always include the meta information parser as the first link in the chain
-            metaState->setNextLink(*currentVisitor);
-            visitIterableEvents(*input, *metaState);
+            Owned<IEventVisitationLink> metaCollector = metaState->getCollector();
+            metaCollector->setNextLink(*currentVisitor);
+            visitIterableEvents(*input, *metaCollector);
             return true;
         }
 

dali/ft/daftformat.cpp

@@ -143,7 +143,9 @@ void CPartitioner::commonCalcPartitions()
         //Don't add an empty block on the start of the this chunk to transfer.
         if ((split != firstSplit) || (inputOffset != startInputOffset))
         {
-            results.append(*new PartitionPoint(whichInput, split-1, startInputOffset-thisOffset+thisHeaderSize, inputOffset - startInputOffset - cursor.trimLength, cursor.outputOffset-startOutputOffset));
+            offset_t startInputOffsetWithTrim = startInputOffset + cursor.trimLength;
+            assertex(startInputOffsetWithTrim <= inputOffset); // guard against length underflow
+            results.append(*new PartitionPoint(whichInput, split-1, startInputOffset-thisOffset+thisHeaderSize, inputOffset - startInputOffsetWithTrim, cursor.outputOffset-startOutputOffset));
             JSON_DBGLOG("commonCalcPartitions: startInputOffset:%lld, inputOffset: %lld, startOutputOffset: %lld, cursor.outputOffset: %lld",startInputOffset ,inputOffset, startOutputOffset, cursor.outputOffset);
             startInputOffset = inputOffset;
             startOutputOffset = cursor.outputOffset;

dali/ft/daftformat.ipp

@@ -563,11 +563,13 @@ protected:
         JSON_DBGLOG("CJsonInputPartitioner::findSplitPoint: splitOffset %lld", splitOffset);
         JSON_DBGLOG("CJsonInputPartitioner::findSplitPoint: cursor(inputOffset: %lld, nextInputOffset: %lld, outputOffset: %lld, trimLength: %lld",
                         cursor.inputOffset, cursor.nextInputOffset, cursor.outputOffset, cursor.trimLength);
-        cursor.inputOffset = 0;
         if (!splitOffset) //header + 0 is first offset
+        {
+            cursor.inputOffset = 0;
             return;
+        }
 
-        if (eof)
+        if (eof) // leave cursor.inputOffset untouched
             return;
 
         // To prevent the splitOffset points into an already processed file area
@@ -579,15 +581,12 @@ protected:
         bool foundRowEnd = json->findRowEnd(splitOffset-thisOffset + thisHeaderSize, prevRowEnd);
         JSON_DBGLOG("CJsonInputPartitioner::findSplitPoint: thisOffset: %lld, thisHeaderSize: %u, prevRowEnd: %lld, foundRowEnd:%s ", thisOffset, thisHeaderSize, prevRowEnd, (foundRowEnd ? "True" : "False"));
         if (! foundRowEnd) //false return just means we're processing the end
-        {
-            //cursor.inputOffset = prevRowEnd;
-            return;
-        }
+            return; // leave cursor.inputOffset untouched
 
         bool checkFoundRowStartRes = json->checkFoundRowStart();
         JSON_DBGLOG("CJsonInputPartitioner::findSplitPoint: checkFoundRowStartRes:%s)", (checkFoundRowStartRes ? "True" : "False"));
         if (!checkFoundRowStartRes)
-            return;
+            return; // leave cursor.inputOffset untouched
 
         bool isNewRowSet = json->newRowSet;
         JSON_DBGLOG("CJsonInputPartitioner::findSplitPoint: isNewRowSet:%s", (isNewRowSet ? "True" : "False"));