HPCC-18622 Store payload information in RtlTypeInfo structures

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

common/deftype/deftype.cpp

@@ -1915,6 +1915,17 @@ static ITypeInfo * commonUpType(CHashedTypeInfo * candidate)
     return match;
 }
 
+extern DEFTYPE_API ITypeInfo *makeFilePosType(ITypeInfo *basetype)
+{
+    assertex(basetype);
+    return commonUpType(new CFilePosTypeInfo(basetype));
+}
+
+extern DEFTYPE_API ITypeInfo *makeKeyedType(ITypeInfo *basetype)
+{
+    assertex(basetype);
+    return commonUpType(new CKeyedTypeInfo(basetype));
+}
 
 extern DEFTYPE_API ITypeInfo *makeDecimalType(unsigned digits, unsigned prec, bool isSigned)
 {

common/deftype/deftype.hpp

@@ -170,6 +170,8 @@ extern DEFTYPE_API ITypeInfo *makeIntType(int size, bool isSigned);
 extern DEFTYPE_API ITypeInfo *makeSwapIntType(int size, bool isSigned);
 extern DEFTYPE_API ITypeInfo *makePackedIntType(ITypeInfo * basetype);
 extern DEFTYPE_API ITypeInfo *makePackedIntType(int size, bool isSigned);
+extern DEFTYPE_API ITypeInfo *makeFilePosType(ITypeInfo *basetype);
+extern DEFTYPE_API ITypeInfo *makeKeyedType(ITypeInfo *basetype);
 extern DEFTYPE_API ITypeInfo *makeRealType(int size);
 extern DEFTYPE_API ITypeInfo *makeDataType(int size);
 extern DEFTYPE_API ITypeInfo *makeBitfieldType(int sizeInBits, ITypeInfo * basetype = NULL);

common/deftype/deftype.ipp

@@ -559,6 +559,43 @@ protected:
     Owned<ITypeInfo>    basetype;
 };
 
+class CFilePosTypeInfo : public CBasedTypeInfo
+{
+public:
+    CFilePosTypeInfo(ITypeInfo * _basetype) : CBasedTypeInfo(_basetype, 0) {}
+    virtual type_t getTypeCode() const { return type_filepos; };
+
+    // Only used for generation of type information so no need to fully implement these
+    virtual bool isInteger()                    { return true; };
+    virtual bool isScalar()                     { return true; }
+    virtual bool isSigned()                     { return basetype->isSigned(); }
+    virtual unsigned getStringLen()             { return basetype->getStringLen(); }
+    virtual unsigned getDigits()                { return basetype->getDigits(); }
+    virtual const char *queryTypeName()         { return "filepos"; }
+    virtual ITypeInfo * queryPromotedType()     { return basetype->queryPromotedType(); }
+    virtual StringBuffer &getECLType(StringBuffer & out) { return out.append("filepos"); }
+    virtual unsigned getCrc()                   { return basetype->getCrc(); }
+};
+
+class CKeyedTypeInfo : public CBasedTypeInfo
+{
+public:
+    CKeyedTypeInfo(ITypeInfo * _basetype) : CBasedTypeInfo(_basetype, _basetype->getSize()) {}
+    virtual type_t getTypeCode() const { return type_keyedint; };
+
+    // Only used for generation of type information so no need to fully implement these
+    virtual bool isSwappedEndian()              { return true; }
+    virtual bool isInteger()                    { return true; };
+    virtual bool isScalar()                     { return true; }
+    virtual bool isSigned()                     { return basetype->isSigned(); }
+    virtual unsigned getStringLen()             { return basetype->getStringLen(); }
+    virtual unsigned getDigits()                { return basetype->getDigits(); }
+    virtual const char *queryTypeName()         { return "keyed"; }
+    virtual ITypeInfo * queryPromotedType()     { return basetype->queryPromotedType(); }
+    virtual StringBuffer &getECLType(StringBuffer & out) { return out.append("keyed"); }
+    virtual unsigned getCrc()                   { return basetype->getCrc(); }
+};
+
 class CPackedIntTypeInfo : public CBasedTypeInfo
 {
 public:

ecl/hql/hqlir.cpp

@@ -1425,6 +1425,7 @@ protected:
                 line.append("u");
             line.append(irText);
             return;
+        case type_filepos:
         case type_int:
         case type_swapint:
             {
@@ -1863,9 +1864,11 @@ id_t ExpressionIRPlayer::doProcessType(ITypeInfo * type)
         case type_event:
         case type_null:
         case type_void:
+        case type_filepos:
         case type_sortlist:
         case type_any:
             break;
+        case type_keyedint:
         case type_int:
         case type_swapint:
         case type_real:

ecl/hql/hqlthql.cpp

@@ -3012,6 +3012,9 @@ StringBuffer &HqltHql::getTypeString(ITypeInfo * i, StringBuffer &s)
     case type_table:
 //  case type_groupedtable:
         break;
+    case type_filepos:
+        getTypeString(i->queryChildType(), s);
+        break;
     default:
         ITypeInfo * original = queryModifier(i, typemod_original);
         IHqlExpression * originalExpr = NULL;

ecl/hql/hqlutil.cpp

@@ -6474,7 +6474,7 @@ IHqlExpression *notePayloadFields(IHqlExpression *record, unsigned payloadCount)
             break;
         case no_field:
         case no_ifblock:
-            fields.replace(*appendOwnedOperand(cur, createAttribute(_payload_Atom)), idx);
+            fields.replace(*appendOwnedOperand(cur, createAttribute(_payload_Atom)), idx); // MORE - should we mark contained fields too?
             payloadCount--;
             break;
         }
@@ -9669,6 +9669,16 @@ void getFieldTypeInfo(FieldTypeInfoStruct &out, ITypeInfo *type)
         if (!type->isSigned())
             out.fieldType |= RFTMunsigned;
         break;
+    case type_keyedint:
+        out.className = "RtlKeyedIntTypeInfo";
+        if (!type->isSigned())
+            out.fieldType |= RFTMunsigned;
+        break;
+    case type_filepos:
+        out.className = "RtlFileposTypeInfo";
+        if (!type->isSigned())
+            out.fieldType |= RFTMunsigned;
+        break;
     case type_swapint:
         out.className = "RtlSwapIntTypeInfo";
         if (!type->isSigned())
@@ -9940,7 +9950,9 @@ const RtlTypeInfo *buildRtlType(IRtlFieldTypeDeserializer &deserializer, ITypeIn
         }
     case type_dictionary:
         return nullptr;  // MORE - does this leak?
+    case type_filepos:
     case type_set:
+    case type_keyedint:
         info.childType = buildRtlType(deserializer, type->queryChildType());
         break;
     }

ecl/hqlcpp/hqlckey.cpp

@@ -1259,7 +1259,7 @@ void KeyedJoinInfo::splitFilter(IHqlExpression * filter, SharedHqlExpr & keyTarg
 
 void HqlCppTranslator::buildKeyedJoinExtra(ActivityInstance & instance, IHqlExpression * expr, KeyedJoinInfo * info)
 {
-    //virtual IOutputMetaData * queryDiskRecordSize() = 0;  // Excluding fpos and sequence
+    //virtual IOutputMetaData * queryDiskRecordSize() = 0;
     if (info->isFullJoin())
         buildMetaMember(instance.classctx, info->queryRawRhs(), false, "queryDiskRecordSize");
 
@@ -1316,8 +1316,18 @@ void HqlCppTranslator::buildKeyJoinIndexReadHelper(ActivityInstance & instance,
     //virtual const char * getIndexFileName() = 0;
     buildFilenameFunction(instance, instance.startctx, "getIndexFileName", info->queryKeyFilename(), hasDynamicFilename(info->queryKey()));
 
-    //virtual IOutputMetaData * queryIndexRecordSize() = 0; //Excluding fpos and sequence
-    buildMetaMember(instance.classctx, info->queryRawKey(), false, "queryIndexRecordSize");
+    //virtual IOutputMetaData * queryIndexRecordSize() = 0;
+    LinkedHqlExpr indexExpr = info->queryOriginalKey();
+    OwnedHqlExpr serializedRecord;
+    if (indexExpr->hasAttribute(_payload_Atom))
+        serializedRecord.setown(notePayloadFields(indexExpr->queryRecord(), numPayloadFields(indexExpr)));
+    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

@@ -3704,10 +3704,9 @@ unsigned HqlCppTranslator::buildRtlField(StringBuffer & instanceName, IHqlExpres
             if (checkXpathIsNonScalar(xpathName))
                 fieldFlags |= RFTMhasnonscalarxpath;
         }
-        if (isPayload)
-        {
+        if (isPayload || field->hasAttribute(_payload_Atom))
             fieldFlags |= RFTMispayloadfield;
-        }
+
         StringBuffer lowerName;
         lowerName.append(field->queryName()).toLowerCase();
 
@@ -3953,7 +3952,9 @@ unsigned HqlCppTranslator::buildRtlType(StringBuffer & instanceName, ITypeInfo *
             arguments.append(",&").append(lookupHelperName.str());
             break;
         }
+    case type_filepos:
     case type_set:
+    case type_keyedint:
         arguments.append(",&");
         childType = buildRtlType(arguments, type->queryChildType());
         break;
@@ -9959,6 +9960,7 @@ void HqlCppTranslator::buildRecordEcl(BuildCtx & subctx, IHqlExpression * record
 void HqlCppTranslator::buildFormatCrcFunction(BuildCtx & ctx, const char * name, IHqlExpression * dataset, IHqlExpression * expr, unsigned payloadDelta)
 {
     IHqlExpression * payload = expr ? expr->queryAttribute(_payload_Atom) : NULL;
+    // MORE - do we need to keep this consistent - if so will have to trim out the originals and the filepos
     OwnedHqlExpr exprToCrc = getSerializedForm(dataset->queryRecord(), diskAtom);
 
     unsigned payloadSize = 1;
@@ -10132,8 +10134,6 @@ void HqlCppTranslator::doBuildIndexOutputTransform(BuildCtx & ctx, IHqlExpressio
         buildReturnRecordSize(func.ctx, selfCursor);
     }
 
-    buildMetaMember(ctx, tgtDataset, false, "queryDiskRecordSize");
-
     size32_t maxRecordSize = 32767;
     if (isVariableSizeRecord(newRecord))
     {
@@ -10333,6 +10333,10 @@ ABoundActivity * HqlCppTranslator::doBuildActivityOutputIndex(BuildCtx & ctx, IH
     serializedRecord.setown(getSerializedForm(serializedRecord, diskAtom));
     buildRecordEcl(instance->createctx, serializedRecord, "queryRecordECL");
 
+    bool hasFilePosition = getBoolAttribute(expr, filepositionAtom, true);
+    serializedRecord.setown(createMetadataIndexRecord(serializedRecord, hasFilePosition));
+    buildMetaMember(instance->classctx, serializedRecord, false, "queryDiskRecordSize");
+
     doBuildSequenceFunc(instance->classctx, querySequence(expr), false);
     HqlExprArray xmlnsAttrs;
     gatherAttributes(xmlnsAttrs, xmlnsAtom, expr);

ecl/hqlcpp/hqlsource.cpp

@@ -471,7 +471,7 @@ static IHqlExpression * mapIfBlock(HqlMapTransformer & mapper, IHqlExpression *
 }
 
 
-static IHqlExpression * createPhysicalIndexRecord(HqlMapTransformer & mapper, IHqlExpression * record, bool hasInternalFileposition)
+static IHqlExpression * createPhysicalIndexRecord(HqlMapTransformer & mapper, IHqlExpression * record, bool hasInternalFileposition, bool createKeyedTypes)
 {
     HqlExprArray physicalFields;
     unsigned max = record->numChildren() - (hasInternalFileposition ? 1 : 0);
@@ -485,7 +485,7 @@ static IHqlExpression * createPhysicalIndexRecord(HqlMapTransformer & mapper, IH
         else if (cur->getOperator() == no_ifblock)
             physicalFields.append(*mapIfBlock(mapper, cur));
         else if (cur->getOperator() == no_record)
-            physicalFields.append(*createPhysicalIndexRecord(mapper, cur, false));
+            physicalFields.append(*createPhysicalIndexRecord(mapper, cur, false, createKeyedTypes));
         else if (cur->hasAttribute(blobAtom))
         {
             newField = createField(cur->queryId(), makeIntType(8, false), NULL, NULL);
@@ -504,8 +504,10 @@ static IHqlExpression * createPhysicalIndexRecord(HqlMapTransformer & mapper, IH
                 Owned<ITypeInfo> hozedType = getHozedKeyType(cur);
                 if (hozedType == cur->queryType())
                     newField = LINK(cur);
+                else if (createKeyedTypes)
+                    newField = createField(cur->queryId(), makeKeyedType(hozedType.getClear()), nullptr, extractFieldAttrs(cur));
                 else
-                    newField = createField(cur->queryId(), hozedType.getClear(), extractFieldAttrs(cur));
+                    newField = createField(cur->queryId(), hozedType.getClear(), nullptr, extractFieldAttrs(cur));
             }
         }
 
@@ -522,10 +524,23 @@ static IHqlExpression * createPhysicalIndexRecord(HqlMapTransformer & mapper, IH
             }
         }
     }
+    if (hasInternalFileposition && createKeyedTypes)
+    {
+        IHqlExpression * cur = record->queryChild(record->numChildren()-1);
+        IHqlExpression *fposField = createField(cur->queryId(), makeFilePosType(cur->getType()), nullptr, extractFieldAttrs(cur));
+        physicalFields.append(*fposField);
+    }
 
     return createRecord(physicalFields);
 }
 
+IHqlExpression * createMetadataIndexRecord(IHqlExpression * record, bool hasInternalFilePosition)
+{
+    HqlMapTransformer mapper;
+    return createPhysicalIndexRecord(mapper, record, hasInternalFilePosition, true);
+}
+
+
 IHqlExpression * HqlCppTranslator::convertToPhysicalIndex(IHqlExpression * tableExpr)
 {
     LinkedHqlExpr * match = physicalIndexCache.getValue(tableExpr);
@@ -540,7 +555,7 @@ IHqlExpression * HqlCppTranslator::convertToPhysicalIndex(IHqlExpression * table
 
     HqlMapTransformer mapper;
     bool hasFileposition = getBoolAttribute(tableExpr, filepositionAtom, true);
-    IHqlExpression * diskRecord = createPhysicalIndexRecord(mapper, record, hasFileposition);
+    IHqlExpression * diskRecord = createPhysicalIndexRecord(mapper, record, hasFileposition, false);
 
     unsigned payload = numPayloadFields(tableExpr);
     assertex(payload || !hasFileposition);
@@ -1985,14 +2000,39 @@ ABoundActivity * SourceBuilder::buildActivity(BuildCtx & ctx, IHqlExpression * e
 
         if (tableExpr && (activityKind < TAKchildread || activityKind > TAKchildthroughnormalize))
         {
-            if (fieldInfo.hasVirtualsOrDeserialize())
+            switch (activityKind)
             {
-                OwnedHqlExpr diskTable = createDataset(no_anon, LINK(physicalRecord));
-                translator.buildMetaMember(instance->classctx, diskTable, false, "queryDiskRecordSize");
-            }
-            else
-                translator.buildMetaMember(instance->classctx, tableExpr, isGrouped(tableExpr), "queryDiskRecordSize");
+            case TAKindexread:
+            case TAKindexnormalize:
+            case TAKindexaggregate:
+            case TAKindexcount:
+            case TAKindexgroupaggregate:
+            case TAKindexexists:
+            {
+                LinkedHqlExpr indexExpr = queryAttributeChild(tableExpr, _original_Atom, 0);
+                OwnedHqlExpr serializedRecord;
+                if (indexExpr->hasAttribute(_payload_Atom))
+                    serializedRecord.setown(notePayloadFields(indexExpr->queryRecord(), numPayloadFields(indexExpr)));
+                else
+                    serializedRecord.set(indexExpr->queryRecord());
+                serializedRecord.setown(getSerializedForm(serializedRecord, diskAtom));
 
+                bool hasFilePosition = getBoolAttribute(indexExpr, filepositionAtom, true);
+                serializedRecord.setown(createMetadataIndexRecord(serializedRecord, hasFilePosition));
+                translator.buildMetaMember(instance->classctx, serializedRecord, false, "queryDiskRecordSize");
+                break;
+            }
+            default:
+                if (fieldInfo.hasVirtualsOrDeserialize())
+                {
+                    OwnedHqlExpr diskTable = createDataset(no_anon, LINK(physicalRecord));
+                    translator.buildMetaMember(instance->classctx, diskTable, false, "queryDiskRecordSize");
+                }
+                else
+                {
+                    translator.buildMetaMember(instance->classctx, tableExpr, isGrouped(tableExpr), "queryDiskRecordSize");
+                }
+            }
         }
     }
     else

ecl/hqlcpp/hqlsource.ipp

@@ -274,5 +274,6 @@ public:
 //---------------------------------------------------------------------------
 
 unsigned getProjectCount(IHqlExpression * expr);
+IHqlExpression * createMetadataIndexRecord(IHqlExpression * record, bool hasInternalFilePosition);
 
 #endif

ecl/hqlcpp/hqlwcpp.cpp

@@ -515,6 +515,11 @@ void HqlCppWriter::generateType(ITypeInfo * type, const char * name)
         case type_boolean:
             prefix = "bool";
             break;
+        case type_filepos:
+            prefix = intTypeName(8-1, compiler, type->isSigned());
+            break;
+        case type_keyedint:
+            break;
         case type_int:
         case type_swapint:
             prefix = intTypeName(size-1, compiler, type->isSigned());

roxie/ccd/ccdactivities.cpp

@@ -3109,7 +3109,7 @@ public:
     void init(IHThorIndexReadBaseArg * helper, IPropertyTree &graphNode)
     {
         rtlDataAttr indexLayoutMeta;
-        size32_t indexLayoutSize;
+        size32_t indexLayoutSize = 0;
         if(!helper->getIndexLayout(indexLayoutSize, indexLayoutMeta.refdata()))
             assertex(indexLayoutSize== 0);
         MemoryBuffer m;
@@ -4634,7 +4634,7 @@ public:
     {
         Owned<IHThorKeyedJoinArg> helper = (IHThorKeyedJoinArg *) helperFactory();
         rtlDataAttr indexLayoutMeta;
-        size32_t indexLayoutSize;
+        size32_t indexLayoutSize = 0;
         if(!helper->getIndexLayout(indexLayoutSize, indexLayoutMeta.refdata()))
             assertex(indexLayoutSize== 0);
         MemoryBuffer m;

roxie/ccd/ccdserver.cpp

@@ -26448,7 +26448,7 @@ public:
         isLocal = _graphNode.getPropBool("att[@name='local']/@value") && queryFactory.queryChannel()!=0;
         isSimple = isLocal;
         rtlDataAttr indexLayoutMeta;
-        size32_t indexLayoutSize;
+        size32_t indexLayoutSize = 0;
         if(!helper->getIndexLayout(indexLayoutSize, indexLayoutMeta.refdata()))
             assertex(indexLayoutSize== 0);
         MemoryBuffer m;

rtl/eclrtl/rtlfield.hpp

@@ -138,6 +138,9 @@ struct ECLRTL_API RtlFileposTypeInfo : public RtlTypeInfoBase
     constexpr inline RtlFileposTypeInfo(unsigned _fieldType, unsigned _length, const RtlTypeInfo *_child, IThorIndexCallback *_callback)
     : RtlTypeInfoBase(_fieldType, _length), child(_child), callback(_callback)
     {}
+    constexpr inline RtlFileposTypeInfo(unsigned _fieldType, unsigned _length, const RtlTypeInfo *_child)
+    : RtlTypeInfoBase(_fieldType, _length), child(_child), callback(nullptr)
+    {}
     virtual void doDelete() const final override { delete this; }
 
     virtual size32_t getMinSize() const override;

rtl/eclrtl/rtlkey.cpp

@@ -810,7 +810,7 @@ public:
     CNewVarOffsetKeySegmentMonitor(IKeySegmentMonitor * _base, unsigned _offset, unsigned _fieldIdx)
     : CIndirectKeySegmentMonitor(_base, 0), fieldIdx(_fieldIdx)
     {
-        assert(_offset = 0);   // We no longer use partial size for offset
+        assert(_offset == 0);   // We no longer use partial size for offset
     }
 
     CNewVarOffsetKeySegmentMonitor(MemoryBuffer &mb)

rtl/include/eclhelper.hpp

@@ -1693,7 +1693,7 @@ struct IHThorKeyedJoinBaseArg : public IHThorArg
 
     // Inside the indexRead remote activity:
     virtual const char * getIndexFileName() = 0;
-    virtual IOutputMetaData * queryIndexRecordSize() = 0; //Excluding fpos and sequence
+    virtual IOutputMetaData * queryIndexRecordSize() = 0;
     virtual void createSegmentMonitors(IIndexReadContext *ctx, const void *lhs) = 0;
     virtual bool indexReadMatch(const void * indexRow, const void * inputRow, unsigned __int64 keyedFpos, IBlobProvider * blobs) = 0;
     virtual unsigned getJoinLimit() = 0;                                        // if a key joins more than this limit no records are output (0 = no limit)