Merge pull request #10682 from richardkchapman/separate-wildmonitors

HPCC-18776 Fix steplimit2 test

Reviewed-by: Gavin Halliday <ghalliday@hpccsystems.com>

common/thorhelper/layouttrans.cpp

@@ -555,11 +555,6 @@ void ExpandedSegmentMonitorList::append(IKeySegmentMonitor * monitor)
     monitors.append(*monitor);
 }
 
-void ExpandedSegmentMonitorList::setMergeBarrier(unsigned offset)
-{
-    // MORE - It's possible that I need to do something here??
-}
-
 IRecordLayoutTranslator * createRecordLayoutTranslator(IDefRecordMeta const * diskMeta, IDefRecordMeta const * activityMeta, IRecordLayoutTranslator::Mode _mode)
 {
     Owned<IRecordLayoutTranslator> layoutTrans = new CRecordLayoutTranslator(diskMeta, activityMeta, _mode);

common/thorhelper/layouttrans.ipp

@@ -99,7 +99,6 @@ class ExpandedSegmentMonitorList : public IRecordLayoutTranslator::SegmentMonito
 public:
     ExpandedSegmentMonitorList(CRecordLayoutTranslator * _owner) : owner(_owner) {}
     IMPLEMENT_IINTERFACE;
-    virtual void setMergeBarrier(unsigned barrierOffset);
     virtual void append(IKeySegmentMonitor * monitor);
     virtual bool isLastSegmentWild() const { return false; }
     virtual unsigned ordinality() const { return monitors.ordinality(); }

ecl/hqlcpp/hqlckey.cpp

@@ -575,8 +575,6 @@ void KeyedJoinInfo::buildLeftOnly(BuildCtx & ctx)
 
 void KeyedJoinInfo::buildMonitors(BuildCtx & ctx)
 {
-    monitors->optimizeSegments(keyAccessDataset->queryRecord());
-
     //---- virtual void createSegmentMonitors(struct IIndexReadContext *) { ... } ----
     MemberFunction func(translator, ctx, "virtual void createSegmentMonitors(IIndexReadContext *irc, const void * _left) override");
     func.ctx.addQuotedLiteral("const unsigned char * left = (const unsigned char *) _left;");
@@ -1306,8 +1304,9 @@ void HqlCppTranslator::buildKeyJoinIndexReadHelper(ActivityInstance & instance,
     //virtual IOutputMetaData * queryIndexRecordSize() = 0;
     LinkedHqlExpr indexExpr = info->queryOriginalKey();
     OwnedHqlExpr serializedRecord;
-    if (indexExpr->hasAttribute(_payload_Atom))
-        serializedRecord.setown(notePayloadFields(indexExpr->queryRecord(), numPayloadFields(indexExpr)));
+    unsigned numPayload = numPayloadFields(indexExpr);
+    if (numPayload)
+        serializedRecord.setown(notePayloadFields(indexExpr->queryRecord(), numPayload));
     else
         serializedRecord.set(indexExpr->queryRecord());
     serializedRecord.setown(getSerializedForm(serializedRecord, diskAtom));
@@ -1559,8 +1558,19 @@ ABoundActivity * HqlCppTranslator::doBuildActivityKeyedDistribute(BuildCtx & ctx
     //virtual const char * getIndexFileName() = 0;
     buildFilenameFunction(*instance, instance->startctx, "getIndexFileName", keyFilename, dynamic);
 
-    //virtual IOutputMetaData * queryIndexRecordSize() = 0; //Excluding fpos and sequence
-    buildMetaMember(instance->classctx, info.queryRawKey(), false, "queryIndexRecordSize");
+    //virtual IOutputMetaData * queryIndexRecordSize() = 0;
+    LinkedHqlExpr indexExpr = info.queryRawKey();
+    OwnedHqlExpr serializedRecord;
+    unsigned numPayload = numPayloadFields(indexExpr);
+    if (numPayload)
+        serializedRecord.setown(notePayloadFields(indexExpr->queryRecord(), numPayload));
+    else
+        serializedRecord.set(indexExpr->queryRecord());
+    serializedRecord.setown(getSerializedForm(serializedRecord, diskAtom));
+
+    bool hasFilePosition = getBoolAttribute(indexExpr, filepositionAtom, true);
+    serializedRecord.setown(createMetadataIndexRecord(serializedRecord, hasFilePosition));
+    buildMetaMember(instance->classctx, serializedRecord, false, "queryIndexRecordSize");
 
     //virtual void createSegmentMonitors(IIndexReadContext *ctx, const void *lhs) = 0;
     info.buildMonitors(instance->startctx);

ecl/hqlcpp/hqlhtcpp.cpp

@@ -10320,8 +10320,6 @@ ABoundActivity * HqlCppTranslator::doBuildActivityOutputIndex(BuildCtx & ctx, IH
     buildUpdateHelper(instance->createctx, *instance, dataset, updateAttr);
     buildClusterHelper(instance->classctx, expr);
 
-    LinkedHqlExpr serializedRecord = record;
-
     // virtual unsigned getKeyedSize()
     HqlExprArray fields;
     unwindChildren(fields, record);
@@ -10329,14 +10327,15 @@ ABoundActivity * HqlCppTranslator::doBuildActivityOutputIndex(BuildCtx & ctx, IH
     fields.popn(numPayloadFields(expr));
     OwnedHqlExpr keyedRecord = createRecord(fields); // must be fixed length => no maxlength
     if (expr->hasAttribute(_payload_Atom))
-    {
         instance->classctx.addQuoted(s.clear().append("virtual unsigned getKeyedSize() override { return ").append(getFixedRecordSize(keyedRecord)).append("; }"));
-        serializedRecord.setown(notePayloadFields(serializedRecord, numPayloadFields(expr)));
-    }
     else
         instance->classctx.addQuoted(s.clear().append("virtual unsigned getKeyedSize() override { return (unsigned) -1; }"));
 
     //virtual const char * queryRecordECL() = 0;
+    LinkedHqlExpr serializedRecord = record;
+    unsigned numPayload = numPayloadFields(expr);
+    if (numPayload)
+        serializedRecord.setown(notePayloadFields(serializedRecord, numPayload));
     serializedRecord.setown(getSerializedForm(serializedRecord, diskAtom));
     buildRecordEcl(instance->createctx, serializedRecord, "queryRecordECL");
 

ecl/hqlcpp/hqlsource.cpp

@@ -483,7 +483,7 @@ static IHqlExpression * createPhysicalIndexRecord(HqlMapTransformer & mapper, IH
             physicalFields.append(*createPhysicalIndexRecord(mapper, cur, false, createKeyedTypes));
         else if (cur->hasAttribute(blobAtom))
         {
-            newField = createField(cur->queryId(), makeIntType(8, false), NULL, NULL);
+            newField = createField(cur->queryId(), makeIntType(8, false), nullptr, createKeyedTypes ? createAttribute(_payload_Atom) : nullptr);
         }
         else
         {
@@ -2014,8 +2014,9 @@ ABoundActivity * SourceBuilder::buildActivity(BuildCtx & ctx, IHqlExpression * e
             {
                 LinkedHqlExpr indexExpr = queryAttributeChild(tableExpr, _original_Atom, 0);
                 OwnedHqlExpr serializedRecord;
-                if (indexExpr->hasAttribute(_payload_Atom))
-                    serializedRecord.setown(notePayloadFields(indexExpr->queryRecord(), numPayloadFields(indexExpr)));
+                unsigned numPayload = numPayloadFields(indexExpr);
+                if (numPayload)
+                    serializedRecord.setown(notePayloadFields(indexExpr->queryRecord(), numPayload));
                 else
                     serializedRecord.set(indexExpr->queryRecord());
                 serializedRecord.setown(getSerializedForm(serializedRecord, diskAtom));
@@ -4781,8 +4782,6 @@ void MonitorExtractor::buildSegments(BuildCtx & ctx, const char * listName, bool
         //      and field->queryType()->getSize() doesn't work for alien datatypes etc.
         if(!field->hasAttribute(virtualAtom))
         {
-            if (mergedSizes.isItem(idx))
-                curSize = mergedSizes.item(idx);
             if (curSize)
                 buildKeySegment(buildState, ctx, idx, curSize);
             else
@@ -6073,47 +6072,6 @@ IHqlExpression * MonitorExtractor::querySimpleJoinValue(IHqlExpression * selecto
     return matched;
 }
 
-
-void MonitorExtractor::optimizeSegments(IHqlExpression * leftRecord)
-{
-    //loop to see if we have matches for key.cpt[n] = x.field[n] and key.cpt[n+1] = x.field[n+1]
-    //where fields are fixed length, no casts and no modifiers.
-    //if so, mark the new total size,
-    //when generate, extend the size of the first monitor, and skip the others.
-    //MORE: Could also combine fixed constants, but less of an advantage.
-    //Don't process anything after a variable size field/something that is transformed.
-    unsigned i = 0;
-    for (; i < firstOffsetField; i++)
-    {
-        IHqlExpression * keySelector = &keyableSelects.item(i);
-        unsigned mergedSize = keySelector->queryType()->getSize();
-        IHqlExpression * prevValue = querySimpleJoinValue(keySelector);
-        unsigned best = i;
-        if (prevValue && isSameBasicType(keySelector->queryType(), prevValue->queryType()))
-        {
-            for (unsigned j = i+1; j < firstOffsetField; j++)
-            {
-                IHqlExpression * nextSelector = &keyableSelects.item(j);
-                if (noMergeSelects.contains(*nextSelector))
-                    break;
-
-                IHqlExpression * nextValue = querySimpleJoinValue(nextSelector);
-                if (!nextValue || !isNextField(leftRecord, prevValue, nextValue) ||
-                    !isSameBasicType(nextSelector->queryType(), nextValue->queryType()))
-                    break;
-                prevValue = nextValue;
-                mergedSize += nextSelector->queryType()->getSize();
-                best = j;
-            }
-        }
-        mergedSizes.append(mergedSize);
-        for (;i < best;i++)
-            mergedSizes.append(0);
-    }
-    while ( i < numKeyableFields)
-        mergedSizes.append(expandedSelects.item(i).queryType()->getSize());
-}
-
 //---------------------------------------------------------------------------
 
 

ecl/hqlcpp/hqlsource.ipp

@@ -151,7 +151,6 @@ public:
     bool isKeyedExplicitly()                                { return keyedExplicitly; }
     bool isFiltered()                                       { return keyed.postFilter || isKeyed(); }
     bool isKeyed();
-    void optimizeSegments(IHqlExpression * leftRecord);
     IHqlExpression * queryGlobalGuard()                     { return keyed.preFilter; }
     void reportFailureReason(IHqlExpression * cond)         { failReason.reportError(translator, cond); }
     const char * queryKeyName(StringBuffer & s);
@@ -218,7 +217,6 @@ protected:
 //  LinkedHqlExpr filter;
 //  LinkedHqlExpr globalGuard;
     KeyConditionInfo keyed;
-    UnsignedArray mergedSizes;
     unsigned numKeyableFields;
     KeyFailureInfo failReason;
 

ecl/hthor/hthor.cpp

@@ -8391,11 +8391,6 @@ void CHThorBinaryDiskReadBase::append(IKeySegmentMonitor *segment)
     }
 }
 
-void CHThorBinaryDiskReadBase::setMergeBarrier(unsigned barrierOffset)
-{
-    // nothing to do - we don't merge...
-}
-
 unsigned CHThorBinaryDiskReadBase::ordinality() const
 {
     return segMonitors.length();

ecl/hthor/hthor.ipp

@@ -2307,7 +2307,6 @@ public:
     virtual void append(IKeySegmentMonitor *segment);
     virtual unsigned ordinality() const;
     virtual IKeySegmentMonitor *item(unsigned idx) const;
-    virtual void setMergeBarrier(unsigned barrierOffset);
 
 protected:
     virtual void verifyRecordFormatCrc() { ::verifyFormatCrcSuper(helper.getFormatCrc(), ldFile?ldFile->queryDistributedFile():NULL, false, true); }

roxie/ccd/ccdactivities.cpp

@@ -972,11 +972,6 @@ public:
         return cursor ? cursor->item(idx) : 0;
     }
 
-    virtual void setMergeBarrier(unsigned barrierOffset)
-    {
-        // no merging so no issue...
-    }
-
     virtual void abort() 
     {
         CRoxieSlaveActivity::abort();
@@ -4152,7 +4147,6 @@ public:
                 groupSegSize = aggregateHelper->getGroupSegmentMonitorsSize();
             else
                 groupSegSize = 0;
-            tlk->setMergeBarrier(groupSegSize);
             CRoxieIndexActivity::createSegmentMonitors();
             if (groupSegSize)
             {

roxie/ccd/ccdkey.cpp

@@ -291,11 +291,6 @@ public:
         totalCount += segment->getSize();
     }
 
-    virtual void setMergeBarrier(unsigned barrierOffset)
-    {
-        // We don't merge segmonitors so nothing to do
-    }
-
     virtual unsigned ordinality() const
     {
         return segMonitors.length();
@@ -1818,11 +1813,6 @@ public:
         segment->Release();
     }
 
-    virtual void setMergeBarrier(unsigned)
-    {
-        // we don't merge segmonitors - nothing to do
-    }
-
     virtual unsigned ordinality() const
     {
         return segMonitors.length();

roxie/ccd/ccdserver.cpp

@@ -21579,11 +21579,6 @@ public:
         return cursor ? cursor->item(idx) : 0;
     }
 
-    virtual void setMergeBarrier(unsigned barrierOffset)
-    {
-        // no merging so no issue...
-    }
-
     virtual void stop()
     {
         if (useRemote())
@@ -24310,7 +24305,6 @@ public:
             groupSegSize = 0;
         if (groupSegSize)
         {
-            key->setMergeBarrier(groupSegSize);
             CRoxieServerIndexActivity::createSegmentMonitors(key);
             unsigned numSegs = tlk->ordinality();
             for (unsigned segNo = 0; segNo < numSegs; segNo++)

rtl/eclrtl/rtlkey.hpp

@@ -134,7 +134,6 @@ public:
     virtual void append(IKeySegmentMonitor *segment) = 0;
     virtual unsigned ordinality() const = 0;
     virtual IKeySegmentMonitor *item(unsigned idx) const = 0;
-    virtual void setMergeBarrier(unsigned offset) = 0;
 };
 
 ECLRTL_API IStringSet *createRtlStringSet(size32_t size);

rtl/eclrtl/rtlrecord.cpp

@@ -348,6 +348,18 @@ RtlRecord::~RtlRecord()
     delete nameMap;
 }
 
+unsigned RtlRecord::getNumKeyedFields() const
+{
+    unsigned ret = 0;
+    for (const RtlFieldInfo * const * finger = originalFields; *finger; finger++)
+    {
+        if ((*finger)->flags & RFTMispayloadfield)
+            break;
+        ret++;
+    }
+    return ret;
+}
+
 void RtlRecord::calcRowOffsets(size_t * variableOffsets, const void * _row, unsigned numFieldsUsed) const
 {
     const byte * row = static_cast<const byte *>(_row);
@@ -501,6 +513,12 @@ bool RtlRecord::excluded(const RtlFieldInfo *field, const byte *row, byte *condi
     return (field->omitable() && static_cast<const RtlCondFieldStrInfo *>(field)->ifblock.excluded(row, conditionValues));
 }
 
+size_t RtlRecord::getFixedOffset(unsigned field) const
+{
+    assert(whichVariableOffset[field]==0);
+    return fixedOffsets[field];
+}
+
 size32_t RtlRecord::getRecordSize(const void *_row) const
 {
     size32_t size = getFixedSize();

rtl/eclrtl/rtlrecord.hpp

@@ -214,6 +214,7 @@ public:
         return fixedOffsets[field] + variableOffsets[whichVariableOffset[field]];
     }
 
+    size_t getFixedOffset(unsigned field) const;
     size_t getRecordSize(size_t * variableOffsets) const
     {
         return getOffset(variableOffsets, numFields);
@@ -225,6 +226,7 @@ public:
     void readAhead(IRowDeserializerSource & in) const;
 
     inline unsigned getNumFields() const { return numFields; }
+    unsigned getNumKeyedFields() const;
     inline unsigned getNumVarFields() const { return numVarFields; }
     inline unsigned getNumIfBlocks() const { return numIfBlocks > 0; }
     inline const RtlFieldInfo * queryField(unsigned field) const { return fields[field]; }

system/jhtree/jhtree.cpp

@@ -59,6 +59,7 @@
 #include "keybuild.hpp"
 #include "eclhelper_dyn.hpp"
 #include "layouttrans.hpp"
+#include "rtlrecord.hpp"
 
 static std::atomic<CKeyStore *> keyStore(nullptr);
 static unsigned defaultKeyIndexLimit = 200;
@@ -86,6 +87,12 @@ MODULE_EXIT()
 
 //#define DUMP_NODES
 
+SegMonitorList::SegMonitorList(const RtlRecord &_recInfo, bool _needWild) : recInfo(_recInfo), needWild(_needWild)
+{
+    keySegCount = recInfo.getNumKeyedFields();
+    reset();
+}
+
 unsigned SegMonitorList::ordinality() const
 {
     return segMonitors.length();
@@ -214,18 +221,19 @@ unsigned SegMonitorList::lastFullSeg() const
     return ret;
 }
 
-void SegMonitorList::finish(unsigned keyedSize) // MORE - probably need to know the record layout here, if wilds are to be added properly.
+void SegMonitorList::finish(unsigned keyedSize)
 {
     if (modified)
     {
-        size32_t segSize = getSize();
-        if (segSize < mergeBarrier)
+        while (segMonitors.length() < keySegCount)
         {
-            segMonitors.append(*createWildKeySegmentMonitor(0, segSize, mergeBarrier-segSize));  // MORE - field number is needed, but that requires layout info
-            segSize = mergeBarrier;
+            unsigned idx = segMonitors.length();
+            size32_t offset = recInfo.getFixedOffset(idx);
+            size32_t size = recInfo.getFixedOffset(idx+1) - offset;
+            segMonitors.append(*createWildKeySegmentMonitor(idx, offset, size));
         }
-         if (segSize < keyedSize)
-            segMonitors.append(*createWildKeySegmentMonitor(0, segSize, keyedSize-segSize));  // MORE - field number is needed, but that requires layout info
+        size32_t segSize = getSize();
+        assertex(segSize == keyedSize);
         recalculateCache();
         modified = false;
     }
@@ -239,7 +247,7 @@ void SegMonitorList::recalculateCache()
 void SegMonitorList::reset()
 {
     segMonitors.kill();
-    modified = true; mergeBarrier = 0;
+    modified = true;
 }
 
 void SegMonitorList::swapWith(SegMonitorList &other)
@@ -270,36 +278,13 @@ void SegMonitorList::append(IKeySegmentMonitor *segment)
     unsigned fieldIdx = segment->getFieldIdx();
     unsigned offset = segment->getOffset();
     unsigned size = segment->getSize();
-#if 0
-    // Eventually we will want something like this:
     while (segMonitors.length() < fieldIdx)
     {
         unsigned idx = segMonitors.length();
-        IKeySegmentMonitor * before = createWildKeySegmentMonitor(idx, field(idx));
+        size32_t offset = recInfo.getFixedOffset(idx);
+        size32_t size = recInfo.getFixedOffset(idx+1) - offset;
+        segMonitors.append(*createWildKeySegmentMonitor(idx, offset, size));
     }
-#else
-    // But as we don't have field info yet, and we don't care about it yet, this will do
-    unsigned lastOffset = 0;
-    if (segMonitors.length())
-    {
-        unsigned lpos = segMonitors.length()-1;
-        IKeySegmentMonitor &last = segMonitors.item(lpos);
-        lastOffset = last.getOffset() + last.getSize();
-    }
-    assertex(lastOffset <= offset);
-    if (lastOffset < offset && needWild)
-    {
-        if (mergeBarrier && lastOffset < mergeBarrier && offset > mergeBarrier)
-        {
-            segMonitors.append(*createWildKeySegmentMonitor(0, lastOffset, mergeBarrier-lastOffset));
-            segMonitors.append(*createWildKeySegmentMonitor(0, mergeBarrier, offset-mergeBarrier));
-        }
-        else
-        {
-            segMonitors.append(*createWildKeySegmentMonitor(0, lastOffset, offset-lastOffset));
-        }
-    }
-#endif
     segMonitors.append(*segment);
 }
 
@@ -897,11 +882,6 @@ public:
         }
     }
 
-    virtual void setMergeBarrier(unsigned offset)
-    {
-        activitySegs->setMergeBarrier(offset); 
-    }
-
     virtual void setSegmentMonitors(SegMonitorList &segmentMonitors) override
     {
         segs.swapWith(segmentMonitors);
@@ -2384,14 +2364,12 @@ class CKeyMerger : public CKeyLevelManager
 public:
     CKeyMerger(const RtlRecord &_recInfo, IKeyIndexSet *_keyset, unsigned _sortFieldOffset, IContextLogger *_ctx) : CKeyLevelManager(_recInfo, NULL, _ctx), sortFieldOffset(_sortFieldOffset)
     {
-        segs.setMergeBarrier(sortFieldOffset);
         init();
         setKey(_keyset);
     }
 
     CKeyMerger(const RtlRecord &_recInfo, IKeyIndex *_onekey, unsigned _sortFieldOffset, IContextLogger *_ctx) : CKeyLevelManager(_recInfo, NULL, _ctx), sortFieldOffset(_sortFieldOffset)
     {
-        segs.setMergeBarrier(sortFieldOffset);
         init();
         setKey(_onekey);
     }
@@ -3144,23 +3122,19 @@ protected:
     void testKeys(bool variable)
     {
         const char *json = variable ?
-                "{ \"ty1\": { \"fieldType\": 4, \"length\": 7 }, "
-                "  \"ty2\": { \"fieldType\": 4, \"length\": 3 }, "
-                "  \"ty3\": { \"fieldType\": 15, \"length\": 8 }, "
+                "{ \"ty1\": { \"fieldType\": 4, \"length\": 10 }, "
+                "  \"ty2\": { \"fieldType\": 15, \"length\": 8 }, "
                 " \"fieldType\": 13, \"length\": 10, "
                 " \"fields\": [ "
                 " { \"name\": \"f1\", \"type\": \"ty1\", \"flags\": 4 }, "
-                " { \"name\": \"f2\", \"type\": \"ty2\", \"flags\": 4 }, "
-                " { \"name\": \"f3\", \"type\": \"ty3\", \"flags\": 65551 } "  // 0x01000f i.e. payload and blob
+                " { \"name\": \"f3\", \"type\": \"ty2\", \"flags\": 65551 } "  // 0x01000f i.e. payload and blob
                 " ]"
                 "}"
                 :
-                "{ \"ty1\": { \"fieldType\": 4, \"length\": 7 }, "
-                "  \"ty2\": { \"fieldType\": 4, \"length\": 3 }, "
+                "{ \"ty1\": { \"fieldType\": 4, \"length\": 10 }, "
                 " \"fieldType\": 13, \"length\": 10, "
                 " \"fields\": [ "
                 " { \"name\": \"f1\", \"type\": \"ty1\", \"flags\": 4 }, "
-                " { \"name\": \"f2\", \"type\": \"ty2\", \"flags\": 4 } "
                 " ] "
                 "}";
         Owned<IOutputMetaData> meta = createTypeInfoOutputMetaData(json, nullptr);

system/jhtree/jhtree.hpp

@@ -159,13 +159,14 @@ class jhtree_decl SegMonitorList : implements IInterface, implements IIndexReadC
 {
     unsigned _lastRealSeg() const;
     unsigned cachedLRS;
-    unsigned mergeBarrier;
     bool modified;
     bool needWild;
     const RtlRecord &recInfo;
+    unsigned keySegCount;
+
 public:
     IMPLEMENT_IINTERFACE;
-    inline SegMonitorList(const RtlRecord &_recInfo, bool _needWild) : recInfo(_recInfo), needWild(_needWild) { reset(); }
+    SegMonitorList(const RtlRecord &_recInfo, bool _needWild);
     IArrayOf<IKeySegmentMonitor> segMonitors;
 
     void reset();
@@ -178,7 +179,6 @@ public:
     unsigned lastFullSeg() const;
     bool matched(void *keyBuffer, unsigned &lastMatch) const;
     size32_t getSize() const;
-    inline void setMergeBarrier(unsigned offset) { mergeBarrier = offset; }
 
     void checkSize(size32_t keyedSize, char const * keyname);
     void recalculateCache();

thorlcr/activities/diskread/thdiskreadslave.cpp

@@ -112,11 +112,6 @@ public:
             return NULL;
     }
     
-    virtual void setMergeBarrier(unsigned barrierOffset)
-    {
-        // We don't merge them so no issue... afaik
-    }
-
 friend class CDiskRecordPartHandler;
 };