HPCC-18467 HThor support for record translation

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

common/thorhelper/CMakeLists.txt

@@ -76,6 +76,8 @@ set (    SRCS
     )
 
 include_directories ( 
+         ${CMAKE_BINARY_DIR}
+         ${CMAKE_BINARY_DIR}/oss
          ./../../system/security/securesocket 
          ./../../common/remote 
          ./../../system/jhtree 
@@ -89,6 +91,7 @@ include_directories (
          ./../deftype 
          ./../workunit
          ./../../rtl/include 
+         ./../../ecl/hql
          ./../../roxie/roxiemem
          ./../../testing/unittests
          ./../../system/tbb_sm/tbb/include
@@ -102,6 +105,7 @@ install ( TARGETS thorhelper RUNTIME DESTINATION ${EXEC_DIR} LIBRARY DESTINATION
 target_link_libraries ( thorhelper 
          jlib 
          nbcd 
+         hql
          eclrtl 
          roxiemem
          deftype 

common/thorhelper/thorcommon.cpp

@@ -28,6 +28,8 @@
 #include "eclrtl.hpp"
 #include "rtlread_imp.hpp"
 #include "rtlcommon.hpp"
+#include "eclhelper_dyn.hpp"
+#include "hqlexpr.hpp"
 #include <algorithm>
 #ifdef _USE_NUMA
 #include <numa.h>
@@ -1815,3 +1817,29 @@ void bindMemoryToLocalNodes()
     numa_bitmask_free(nodes);
 #endif
 }
+
+extern THORHELPER_API IOutputMetaData *getDaliLayoutInfo(IPropertyTree const &props)
+{
+    bool isGrouped = props.getPropBool("@grouped", false);
+    if (props.hasProp("_rtlType"))
+    {
+        MemoryBuffer layoutBin;
+        props.getPropBin("_rtlType", layoutBin);
+        return createTypeInfoOutputMetaData(layoutBin, isGrouped, nullptr);
+    }
+    else if (props.hasProp("ECL"))
+    {
+        StringBuffer layoutECL;
+        props.getProp("ECL", layoutECL);
+        MultiErrorReceiver errs;
+        Owned<IHqlExpression> expr = parseQuery(layoutECL.str(), &errs);
+        if (errs.errCount() == 0)
+        {
+            MemoryBuffer layoutBin;
+            if (exportBinaryType(layoutBin, expr))
+                return createTypeInfoOutputMetaData(layoutBin, isGrouped, nullptr);
+        }
+    }
+    return nullptr;
+}
+

common/thorhelper/thorcommon.hpp

@@ -619,4 +619,6 @@ extern THORHELPER_API void setProcessAffinity(const char * cpus);
 extern THORHELPER_API void setAutoAffinity(unsigned curProcess, unsigned processPerNode, const char * optNodes);
 extern THORHELPER_API void bindMemoryToLocalNodes();
 
+extern THORHELPER_API IOutputMetaData *getDaliLayoutInfo(IPropertyTree const &props);
+
 #endif // THORHELPER_HPP

common/workunit/pkgimpl.hpp

@@ -124,12 +124,7 @@ protected:
         const char *val = queryEnv("control:enableFieldTranslation");
         if (!val) val = queryEnv("enableFieldTranslation"); // Backward compatibility
         if (val)
-        {
-            if (strieq(val, "payload") || strToBool(val))
-                return RecordTranslationMode::Payload;
-            else
-                return RecordTranslationMode::None;
-        }
+            return getTranslationMode(val);
         else
             return getSysFieldTranslationEnabled();
     }

ecl/eclagent/agentctx.hpp

@@ -20,6 +20,7 @@
 #include "errorlist.h"
 #include "dautils.hpp"
 #include "eclhelper.hpp"
+#include "rtldynfield.hpp"
 #include "workunit.hpp"
 
 #if (ECLAGENT_ERROR_START != 5400 || ECLAGENT_ERROR_END != 5499)
@@ -86,7 +87,7 @@ struct IAgentContext : extends IGlobalCodeContext
 
     virtual ICodeContext *queryCodeContext() = 0;
 
-    virtual IConstWorkUnit *queryWorkUnit() = 0;
+    virtual IConstWorkUnit *queryWorkUnit() const = 0;
     virtual IWorkUnit *updateWorkUnit() const = 0;
     virtual void unlockWorkUnit() = 0;
     
@@ -112,6 +113,7 @@ struct IAgentContext : extends IGlobalCodeContext
 
     virtual IGroup *getHThorGroup(StringBuffer &grpnameout) = 0;
 
+    virtual RecordTranslationMode rltEnabled() const = 0;
     virtual unsigned __int64 queryStopAfter() = 0;
     
     virtual const char *queryWuid() = 0;

ecl/eclagent/eclagent.cpp

@@ -38,6 +38,7 @@
 #include "eclrtl_imp.hpp"
 #include "rtlds_imp.hpp"
 #include "rtlcommon.hpp"
+#include "rtldynfield.hpp"
 #include "workunit.hpp"
 #include "eventqueue.hpp"
 #include "schedulectrl.hpp"
@@ -700,6 +701,17 @@ void EclAgent::abort()
         activeGraph->abort();
 }
 
+RecordTranslationMode EclAgent::rltEnabled() const
+{
+    IConstWorkUnit *wu = queryWorkUnit();
+    SCMStringBuffer val;
+    if(wu->hasDebugValue("layoutTranslationEnabled"))
+        wu->getDebugValue("layoutTranslationEnabled", val);
+    else
+        wu->getDebugValue("hthorLayoutTranslationEnabled", val);
+    return getTranslationMode(val.str());
+}
+
 IConstWUResult *EclAgent::getResult(const char *name, unsigned sequence)
 {
     IConstWorkUnit *w = queryWorkUnit();
@@ -3019,7 +3031,7 @@ char * EclAgent::queryIndexMetaData(char const * lfn, char const * xpath)
     return out.detach();
 }
 
-IConstWorkUnit *EclAgent::queryWorkUnit()
+IConstWorkUnit *EclAgent::queryWorkUnit() const
 {
     return wuRead;
 }

ecl/eclagent/eclagent.ipp

@@ -23,8 +23,8 @@
 #include "deftype.hpp"
 #include "jthread.hpp"
 #include "dllserver.hpp"
+#include "rtldynfield.hpp"
 
-//#include "agentctx.hpp"
 #include "hthor.hpp"
 #include "thorxmlwrite.hpp"
 #include "workflow.hpp"
@@ -155,7 +155,7 @@ public:
     {
         ctx->reportProgress(msg, flags);
     }
-    virtual IConstWorkUnit *queryWorkUnit()
+    virtual IConstWorkUnit *queryWorkUnit() const override
     {
         return ctx->queryWorkUnit();
     }
@@ -233,6 +233,11 @@ public:
 
     virtual void updateWULogfile()                  { return ctx->updateWULogfile(); }
 
+    virtual RecordTranslationMode rltEnabled() const override
+    {
+        return ctx->rltEnabled();
+    }
+
 protected:
     IAgentContext * ctx;
 };
@@ -502,6 +507,7 @@ public:
     virtual IEngineContext *queryEngineContext() { return this; }
     virtual char *getDaliServers();
 
+    virtual RecordTranslationMode rltEnabled() const override;
     unsigned __int64 queryStopAfter() { return stopAfter; }
 
     virtual ISectionTimer * registerTimer(unsigned activityId, const char * name)
@@ -588,8 +594,8 @@ public:
     virtual const char *loadResource(unsigned id);
     virtual ICodeContext *queryCodeContext();
     virtual bool isResult(const char * name, unsigned sequence);
-    virtual unsigned getWorkflowId();// { return workflow->queryCurrentWfid(); }
-    virtual IConstWorkUnit *queryWorkUnit();  // no link
+    virtual unsigned getWorkflowId();
+    virtual IConstWorkUnit *queryWorkUnit() const override;  // no link
     virtual IWorkUnit *updateWorkUnit() const; // links
     virtual void unlockWorkUnit();      
     virtual void reloadWorkUnit();

ecl/hthor/CMakeLists.txt

@@ -79,7 +79,6 @@ target_link_libraries ( hthor
          jlib
          mp
          hrpc
-         hql
          remote
          dalibase
          environment

ecl/hthor/hthor.cpp

@@ -43,12 +43,14 @@
 #include "thorsort.hpp"
 #include "thorparse.ipp"
 #include "thorxmlwrite.hpp"
+#include "thorcommon.hpp"
 #include "jsmartsock.hpp"
 #include "thorstep.hpp"
 #include "eclagent.ipp"
 #include "roxierowbuff.hpp"
 #include "ftbase.ipp"
 #include "rtldynfield.hpp"
+#include "rtlnewkey.hpp"
 
 #define EMPTY_LOOP_LIMIT 1000
 
@@ -7993,6 +7995,8 @@ const void *CHThorChildThroughNormalizeActivity::nextRow()
 CHThorDiskReadBaseActivity::CHThorDiskReadBaseActivity(IAgentContext &_agent, unsigned _activityId, unsigned _subgraphId, IHThorDiskReadBaseArg &_arg, ThorActivityKind _kind) : CHThorActivityBase(_agent, _activityId, _subgraphId, _arg, _kind), helper(_arg)
 {
     helper.setCallback(this);
+    expectedDiskMeta = helper.queryDiskRecordSize();
+    projectedDiskMeta = helper.queryProjectedDiskRecordSize();
 }
 
 CHThorDiskReadBaseActivity::~CHThorDiskReadBaseActivity()
@@ -8077,7 +8081,7 @@ void CHThorDiskReadBaseActivity::resolve()
                 }
                 if((helper.getFlags() & (TDXtemporary | TDXjobtemp)) == 0)
                     agent.logFileAccess(dFile, "HThor", "READ");
-                if(!agent.queryWorkUnit()->getDebugValueBool("skipFileFormatCrcCheck", false) && !(helper.getFlags() & TDRnocrccheck))
+                if(agent.rltEnabled()==RecordTranslationMode::None && !agent.queryWorkUnit()->getDebugValueBool("skipFileFormatCrcCheck", false) && !(helper.getFlags() & TDRnocrccheck))
                     verifyRecordFormatCrc();
             }
         }
@@ -8114,9 +8118,9 @@ void CHThorDiskReadBaseActivity::gatherInfo(IFileDescriptor * fileDesc)
         grouped = ((helper.getFlags() & TDXgrouped) != 0);
     }
 
-    diskMeta.set(helper.queryDiskRecordSize()->querySerializedDiskMeta());
+    actualDiskMeta.set(helper.queryDiskRecordSize()->querySerializedDiskMeta());
     if (grouped)
-        diskMeta.setown(new CSuffixedOutputMeta(+1, diskMeta));
+        actualDiskMeta.setown(new CSuffixedOutputMeta(+1, actualDiskMeta));
     if (outputMeta.isFixedSize())
     {
         recordsize = outputMeta.getFixedSize();
@@ -8205,6 +8209,8 @@ bool CHThorDiskReadBaseActivity::openNext()
               (!dfsParts&&(partNum<ldFile->numParts())))
         {
             IDistributedFilePart * curPart = dfsParts?&dfsParts->query():NULL;
+            IDistributedFile *dFile = ldFile->queryDistributedFile();  // Null for local file usage
+
             unsigned numCopies = curPart?curPart->numCopies():ldFile->numPartCopies(partNum);
             //MORE: Order of copies should be optimized at this point....
             StringBuffer file, filelist;
@@ -8214,8 +8220,62 @@ bool CHThorDiskReadBaseActivity::openNext()
                 unsigned subfile;
                 unsigned lnum;
                 if (superfile->mapSubPart(partNum, subfile, lnum))
+                {
                     logicalFileName.set(subfileLogicalFilenames.item(subfile));
+                    // MORE - need to set dFile = superfile->getSubFilePart(subfile) to support different formats on different file parts
+                }
+            }
+
+            if (dFile)
+            {
+                IPropertyTree &props = dFile->queryAttributes();
+                unsigned thisFormatCrc = props.getPropInt("@formatCrc");
+                if (thisFormatCrc != lastFormatCrc)
+                {
+                    translator.clear();
+                    lastFormatCrc = thisFormatCrc;
+                    if (thisFormatCrc != helper.getFormatCrc() && helper.getFormatCrc() && (helper.getFlags() & TDRnocrccheck) == 0)
+                    {
+                        actualDiskMeta.setown(getDaliLayoutInfo(props));
+                        if (actualDiskMeta)
+                        {
+                            translator.setown(createRecordTranslator(projectedDiskMeta->queryRecordAccessor(true), actualDiskMeta->queryRecordAccessor(true)));
+                            if (translator->needsTranslate())
+                            {
+                                keyedTranslator.setown(createKeyTranslator(actualDiskMeta->queryRecordAccessor(true), expectedDiskMeta->queryRecordAccessor(true)));
+                                if (translator->canTranslate())
+                                {
+                                    if (agent.rltEnabled()==RecordTranslationMode::None)
+                                    {
+    #ifdef _DEBUG
+                                        translator->describe();
+    #endif
+                                        throw MakeStringException(0, "Translatable key layout mismatch reading file %s but translation disabled", logicalFileName.str());
+                                    }
+                                }
+                                else
+                                    throw MakeStringException(0, "Untranslatable key layout mismatch reading file %s", logicalFileName.str());
+                            }
+                            else
+                                translator.clear();  // MORE - could question why the format appeared to mismatch
+                        }
+                        else
+                            throw MakeStringException(0, "Untranslatable key layout mismatch reading file %s - key layout information not found", logicalFileName.str());
+                    }
+                    else
+                    {
+                        actualDiskMeta.set(helper.queryDiskRecordSize()->querySerializedDiskMeta());
+                        if (grouped)
+                            actualDiskMeta.setown(new CSuffixedOutputMeta(+1, actualDiskMeta));
+                    }
+                }
+            }
+            else
+            {
+                translator.clear();
+                keyedTranslator.clear();
             }
+            calcFixedDiskRecordSize();
             for (unsigned copy=0; copy < numCopies; copy++)
             {
                 RemoteFilename rfilename;
@@ -8378,27 +8438,13 @@ void CHThorDiskReadBaseActivity::open()
 
 CHThorBinaryDiskReadBase::CHThorBinaryDiskReadBase(IAgentContext &_agent, unsigned _activityId, unsigned _subgraphId, IHThorDiskReadBaseArg &_arg, IHThorCompoundBaseArg & _segHelper, ThorActivityKind _kind)
 : CHThorDiskReadBaseActivity(_agent, _activityId, _subgraphId, _arg, _kind),
-  segHelper(_segHelper), prefetchBuffer(NULL),
-  recInfo(outputMeta.queryRecordAccessor(true)),  // MORE - is this right - should it be diskMeta->queryRecordAccessor() ?
-  rowInfo(recInfo)
+  segHelper(_segHelper), prefetchBuffer(NULL)
 {
 }
 
 void CHThorBinaryDiskReadBase::calcFixedDiskRecordSize()
 {
-    fixedDiskRecordSize = diskMeta->getFixedSize();
-}
-
-void CHThorBinaryDiskReadBase::append(IKeySegmentMonitor *segment)
-{
-    if (segment->isWild())
-        segment->Release();
-    else
-    {
-        segMonitors.append(*segment);
-        if (segment->numFieldsRequired() > numFieldsRequired)
-            numFieldsRequired = segment->numFieldsRequired();
-    }
+    fixedDiskRecordSize = actualDiskMeta->getFixedSize();
 }
 
 void CHThorBinaryDiskReadBase::append(FFoption option, IFieldFilter * filter)
@@ -8406,37 +8452,14 @@ void CHThorBinaryDiskReadBase::append(FFoption option, IFieldFilter * filter)
     if (filter->isWild())
         filter->Release();
     else
-    {
         fieldFilters.append(*filter);
-        if (filter->queryFieldIndex() > numFieldsRequired)
-            numFieldsRequired = filter->queryFieldIndex();
-    }
-}
-
-unsigned CHThorBinaryDiskReadBase::ordinality() const
-{
-    return segMonitors.length();
-}
-
-IKeySegmentMonitor *CHThorBinaryDiskReadBase::item(unsigned idx) const
-{
-    if (segMonitors.isItem(idx))
-        return &segMonitors.item(idx);
-    else
-        return NULL;
 }
 
 void CHThorBinaryDiskReadBase::ready()      
 { 
     CHThorDiskReadBaseActivity::ready(); 
-    if (!diskMeta)
-        diskMeta.set(outputMeta);
-    segMonitors.kill();
     fieldFilters.kill();
-    numFieldsRequired = 0;
     segHelper.createSegmentMonitors(this);
-    prefetcher.setown(diskMeta->createDiskPrefetcher());
-    deserializer.setown(diskMeta->createDiskDeserializer(agent.queryCodeContext(), activityId));
 }
 
 bool CHThorBinaryDiskReadBase::openNext()
@@ -8450,8 +8473,15 @@ bool CHThorBinaryDiskReadBase::openNext()
             PROGLOG("Disk read falling back to legacy decompression routine");
             //in.setown(createRowCompReadSeq(*inputfileiostream, 0, fixedDiskRecordSize));
         }
+        actualFilter.clear();
+        if (keyedTranslator)
+            keyedTranslator->translate(actualFilter, fieldFilters);
+        else
+            actualFilter.appendFilters(fieldFilters);
 
         //Only one of these will actually be used.
+        prefetcher.setown(actualDiskMeta->createDiskPrefetcher());
+        deserializer.setown(actualDiskMeta->createDiskDeserializer(agent.queryCodeContext(), activityId));
         prefetchBuffer.setStream(inputstream);
         deserializeSource.setStream(inputstream);
         return true;
@@ -8498,7 +8528,7 @@ void CHThorDiskReadActivity::ready()
     outBuilder.setAllocator(rowAllocator);
     eogPending = false;
     lastGroupProcessed = processed;
-    needTransform = helper.needTransform() || segMonitors.length() || fieldFilters.length();
+    needTransform = helper.needTransform() || fieldFilters.length();
     limit = helper.getRowLimit();
     if (helper.getFlags() & TDRlimitskips)
         limit = (unsigned __int64) -1;
@@ -8525,7 +8555,7 @@ const void *CHThorDiskReadActivity::nextRow()
 
     try
     {
-        if (needTransform || grouped)
+        if (needTransform || grouped || translator || keyedTranslator)
         {
             while (!eofseen && ((stopAfter == 0) || ((processed - initialProcessed) < stopAfter)))
             {
@@ -8538,8 +8568,15 @@ const void *CHThorDiskReadActivity::nextRow()
                     const byte * next = prefetchBuffer.queryRow();
                     size32_t sizeRead = prefetchBuffer.queryRowSize();
                     size32_t thisSize;
-                    if (segMonitorsMatch(next))
+                    if (segMonitorsMatch(next)) // NOTE - keyed fields are checked pre-translation
                     {
+                        MemoryBuffer translated;
+                        if (translator)
+                        {
+                            MemoryBufferBuilder aBuilder(translated, 0);
+                            translator->translate(aBuilder, next);
+                            next = reinterpret_cast<const byte *>(translated.toByteArray());
+                        }
                         thisSize = helper.transform(outBuilder.ensureRow(), next);
                     }
                     else
@@ -8575,7 +8612,6 @@ const void *CHThorDiskReadActivity::nextRow()
         }
         else
         {
-            assertex(outputMeta == diskMeta);
             while(!eofseen && ((stopAfter == 0) || (processed - initialProcessed) < stopAfter)) 
             {
                 queryUpdateProgress();
@@ -8819,7 +8855,7 @@ const void *CHThorDiskCountActivity::nextRow()
     if (finished) return NULL;
 
     unsigned __int64 totalCount = 0;
-    if ((segMonitors.ordinality() == 0) && (fieldFilters.ordinality() == 0) && !helper.hasFilter() &&
+    if (fieldFilters.ordinality() == 0 && !helper.hasFilter() &&
         (fixedDiskRecordSize != 0) && !(helper.getFlags() & (TDXtemporary | TDXjobtemp)) &&
         !((helper.getFlags() & TDXcompress) && agent.queryResolveFilesLocally()) )
     {

ecl/hthor/hthor.ipp

@@ -2241,7 +2241,9 @@ protected:
     Owned<IException> saveOpenExc;
     size32_t recordsize;
     size32_t fixedDiskRecordSize;
-    Owned<IOutputMetaData> diskMeta;
+    Owned<IOutputMetaData> actualDiskMeta;
+    IOutputMetaData *expectedDiskMeta;
+    IOutputMetaData *projectedDiskMeta;
     unsigned partNum;
     bool eofseen;
     bool opened;
@@ -2251,13 +2253,16 @@ protected:
     MemoryAttr encryptionkey;
     bool persistent;
     bool grouped;
+    unsigned lastFormatCrc = 0;
     unsigned __int64 localOffset;
     unsigned __int64 offsetOfPart;
     StringBuffer mangledHelperFileName;
     StringAttr logicalFileName;
     StringArray subfileLogicalFilenames;
     Owned<ISuperFileDescriptor> superfile;
-
+    Owned<const IDynamicTransform> translator;
+    Owned<const IKeyTranslator> keyedTranslator;
+    IPointerArrayOf<IOutputMetaData> actualLayouts;  // Do we need to keep more than one?
     void close();
     virtual void open();
     void resolve();
@@ -2298,26 +2303,23 @@ public:
 class CHThorBinaryDiskReadBase : public CHThorDiskReadBaseActivity, implements IIndexReadContext
 {
 protected:
-    IArrayOf<IKeySegmentMonitor> segMonitors;
-    IArrayOf<IFieldFilter> fieldFilters;
+    IConstArrayOf<IFieldFilter> fieldFilters;  // These refer to the expected layout
+    RowFilter actualFilter;               // This refers to the actual disk layout
     IHThorCompoundBaseArg & segHelper;
     Owned<ISourceRowPrefetcher> prefetcher;
     Owned<IOutputRowDeserializer> deserializer;
     CThorContiguousRowBuffer prefetchBuffer;
     CThorStreamDeserializerSource deserializeSource;
-    const RtlRecord &recInfo;
-    RtlDynRow rowInfo;
-    unsigned numFieldsRequired = 0;
 public:
     CHThorBinaryDiskReadBase(IAgentContext &agent, unsigned _activityId, unsigned _subgraphId, IHThorDiskReadBaseArg &_arg, IHThorCompoundBaseArg & _segHelper, ThorActivityKind _kind);
 
     virtual void ready();
 
     //interface IIndexReadContext
-    virtual void append(IKeySegmentMonitor *segment);
-    virtual unsigned ordinality() const;
-    virtual IKeySegmentMonitor *item(unsigned idx) const;
-    virtual void append(FFoption option, IFieldFilter * filter);
+    virtual void append(IKeySegmentMonitor *segment) override { throwUnexpected(); }
+    virtual unsigned ordinality() const override { throwUnexpected(); }
+    virtual IKeySegmentMonitor *item(unsigned idx) const override { throwUnexpected();  }
+    virtual void append(FFoption option, IFieldFilter * filter) override;
 
 protected:
     virtual void verifyRecordFormatCrc() { ::verifyFormatCrcSuper(helper.getFormatCrc(), ldFile?ldFile->queryDistributedFile():NULL, false, true); }
@@ -2328,28 +2330,17 @@ protected:
 
     inline bool segMonitorsMatch(const void * buffer)
     {
-        bool match = true;
-        if (segMonitors || fieldFilters)
+        if (actualFilter.numFilterFields())
         {
-            rowInfo.setRow(buffer, numFieldsRequired);
-            ForEachItemIn(idx, segMonitors)
-            {
-                if (!segMonitors.item(idx).matches(&rowInfo))
-                {
-                    match = false;
-                    break;
-                }
-            }
-            ForEachItemIn(idx2, fieldFilters)
-            {
-                if (!fieldFilters.item(idx2).matches(rowInfo))
-                {
-                    match = false;
-                    break;
-                }
-            }
+            const RtlRecord &actual = actualDiskMeta->queryRecordAccessor(true);
+            unsigned numOffsets = actual.getNumVarFields() + 1;
+            size_t * variableOffsets = (size_t *)alloca(numOffsets * sizeof(size_t));
+            RtlRow row(actual, nullptr, numOffsets, variableOffsets);
+            row.setRow(buffer, 0);  // Use lazy offset calculation
+            return actualFilter.matches(row);
         }
-        return match;
+        else
+            return true;
     }
 
     virtual void calcFixedDiskRecordSize();

ecl/hthor/hthorkey.cpp

@@ -22,10 +22,9 @@
 #include "jqueue.tpp"
 #include "dasess.hpp"
 #include "thorxmlwrite.hpp"
-#include "eclhelper_dyn.hpp"
 #include "thorstep.ipp"
 #include "roxiedebug.hpp"
-#include "hqlexpr.hpp"
+#include "thorcommon.hpp"
 #include "rtldynfield.hpp"
 
 #define MAX_FETCH_LOOKAHEAD 1000
@@ -83,14 +82,6 @@ void enterSingletonSuperfiles(Shared<IDistributedFile> & file)
     }
 }
 
-bool rltEnabled(IConstWorkUnit const * wu)
-{
-    if(wu->hasDebugValue("layoutTranslationEnabled"))
-        return wu->getDebugValueBool("layoutTranslationEnabled", false);
-    else
-        return wu->getDebugValueBool("hthorLayoutTranslationEnabled", false);
-}
-
 static void setProgress(IPropertyTree &node, const char *name, const char *value)
 {
     StringBuffer attr("@");
@@ -684,7 +675,7 @@ const IDynamicTransform * CHThorIndexReadActivityBase::getLayoutTranslator(IDist
     if(agent.queryWorkUnit()->getDebugValueBool("skipFileFormatCrcCheck", false))
         return NULL;
 
-    if(!rltEnabled(agent.queryWorkUnit()))
+    if(agent.rltEnabled() == RecordTranslationMode::None)
     {
         verifyFormatCrc(helper.getFormatCrc(), f, (superIterator ? superName.str() : NULL) , true, true);
         return NULL;
@@ -693,32 +684,17 @@ const IDynamicTransform * CHThorIndexReadActivityBase::getLayoutTranslator(IDist
     if(verifyFormatCrc(helper.getFormatCrc(), f, (superIterator ? superName.str() : NULL) , true, false))
         return NULL;
 
-    Owned<IOutputMetaData> actualFormat;
     IPropertyTree &props = f->queryAttributes();
-    bool isGrouped = props.getPropBool("@grouped", false);
-    if (props.hasProp("_rtlType"))
-    {
-        MemoryBuffer layoutBin;
-        props.getPropBin("_rtlType", layoutBin);
-        actualFormat.setown(createTypeInfoOutputMetaData(layoutBin, isGrouped, nullptr));
-    }
-    else if (props.hasProp("ECL"))
-    {
-        StringBuffer layoutECL;
-        props.getProp("ECL", layoutECL);
-        MultiErrorReceiver errs;
-        Owned<IHqlExpression> expr = parseQuery(layoutECL.str(), &errs);
-        if (errs.errCount() == 0)
-        {
-            MemoryBuffer layoutBin;
-            if (exportBinaryType(layoutBin, expr))
-                actualFormat.setown(createTypeInfoOutputMetaData(layoutBin, isGrouped, nullptr));
-        }
-    }
+    Owned<IOutputMetaData> actualFormat = getDaliLayoutInfo(props);
     if (actualFormat)
     {
         actualLayouts.append(actualFormat.getLink());  // ensure adequate lifespan
-        return createRecordTranslator(helper.queryProjectedDiskRecordSize()->queryRecordAccessor(true), actualFormat->queryRecordAccessor(true));
+        Owned<const IDynamicTransform> payloadTranslator =  createRecordTranslator(helper.queryProjectedDiskRecordSize()->queryRecordAccessor(true), actualFormat->queryRecordAccessor(true));
+        if (!payloadTranslator->canTranslate())
+            throw MakeStringException(0, "Untranslatable key layout mismatch reading index %s", f->queryLogicalName());
+        if (payloadTranslator->keyedTranslated())
+            throw MakeStringException(0, "Untranslatable key layout mismatch reading index %s - keyed fields do not match", f->queryLogicalName());
+        return payloadTranslator.getClear();
     }
     throw MakeStringException(0, "Untranslatable key layout mismatch reading index %s - key layout information not found", f->queryLogicalName());
 }
@@ -1778,14 +1754,14 @@ protected:
     unsigned activityId;
     CachedOutputMetaData const & outputMeta;
     IEngineRowAllocator * rowAllocator;
-    IOutputRowDeserializer * rowDeserializer;
+    ISourceRowPrefetcher * prefetcher;
 public:
-    FetchPartHandlerBase(offset_t _base, offset_t _size, bool _blockcompressed, MemoryAttr &_encryptionkey, unsigned _activityId, CachedOutputMetaData const & _outputMeta, IOutputRowDeserializer * _rowDeserializer, IEngineRowAllocator *_rowAllocator) 
+    FetchPartHandlerBase(offset_t _base, offset_t _size, bool _blockcompressed, MemoryAttr &_encryptionkey, unsigned _activityId, CachedOutputMetaData const & _outputMeta, ISourceRowPrefetcher * _prefetcher, IEngineRowAllocator *_rowAllocator)
         : blockcompressed(_blockcompressed), 
           encryptionkey(_encryptionkey), 
           activityId(_activityId), 
           outputMeta(_outputMeta),
-          rowDeserializer(_rowDeserializer), 
+          prefetcher(_prefetcher),
           rowAllocator(_rowAllocator)
     {
         base = _base;
@@ -1869,8 +1845,8 @@ public:
 class SimpleFetchPartHandlerBase : public FetchPartHandlerBase, public ThreadedPartHandler<FetchRequest>
 {
 public:
-    SimpleFetchPartHandlerBase(IDistributedFilePart *_part, offset_t _base, offset_t _size, IThreadedExceptionHandler *_handler, IThreadPool * _threadPool, bool _blockcompressed, MemoryAttr &_encryptionkey, unsigned _activityId, CachedOutputMetaData const & _outputMeta, IOutputRowDeserializer * _rowDeserializer, IEngineRowAllocator *_rowAllocator) 
-        : FetchPartHandlerBase(_base, _size, _blockcompressed, _encryptionkey, _activityId, _outputMeta, _rowDeserializer, _rowAllocator), 
+    SimpleFetchPartHandlerBase(IDistributedFilePart *_part, offset_t _base, offset_t _size, IThreadedExceptionHandler *_handler, IThreadPool * _threadPool, bool _blockcompressed, MemoryAttr &_encryptionkey, unsigned _activityId, CachedOutputMetaData const & _outputMeta, ISourceRowPrefetcher * _prefetcher, IEngineRowAllocator *_rowAllocator)
+        : FetchPartHandlerBase(_base, _size, _blockcompressed, _encryptionkey, _activityId, _outputMeta, _prefetcher, _rowAllocator),
           ThreadedPartHandler<FetchRequest>(_part, _handler, _threadPool)
     {
     }
@@ -1893,8 +1869,8 @@ private:
 class FlatFetchPartHandler : public SimpleFetchPartHandlerBase
 {
 public:
-    FlatFetchPartHandler(IFlatFetchHandlerCallback & _owner, IDistributedFilePart * _part, offset_t _base, offset_t _size, IThreadedExceptionHandler *_handler, IThreadPool * _threadPool, bool _blockcompressed, MemoryAttr &_encryptionkey, unsigned _activityId, CachedOutputMetaData const & _outputMeta, IOutputRowDeserializer * _rowDeserializer, IEngineRowAllocator *_rowAllocator)
-        : SimpleFetchPartHandlerBase(_part, _base, _size, _handler, _threadPool, _blockcompressed, _encryptionkey, _activityId, _outputMeta, _rowDeserializer, _rowAllocator), 
+    FlatFetchPartHandler(IFlatFetchHandlerCallback & _owner, IDistributedFilePart * _part, offset_t _base, offset_t _size, IThreadedExceptionHandler *_handler, IThreadPool * _threadPool, bool _blockcompressed, MemoryAttr &_encryptionkey, unsigned _activityId, CachedOutputMetaData const & _outputMeta, ISourceRowPrefetcher * _prefetcher, IEngineRowAllocator *_rowAllocator)
+        : SimpleFetchPartHandlerBase(_part, _base, _size, _handler, _threadPool, _blockcompressed, _encryptionkey, _activityId, _outputMeta, _prefetcher, _rowAllocator),
           owner(_owner)
     {
     }
@@ -1973,7 +1949,7 @@ template <class PARTHANDLER>
 class IFetchHandlerFactory
 {
 public:
-    virtual PARTHANDLER * createFetchPartHandler(IDistributedFilePart * part, offset_t base, offset_t size, IThreadedExceptionHandler * handler, bool blockcompressed, MemoryAttr &encryptionkey, IOutputRowDeserializer * rowDeserializer, IEngineRowAllocator *rowAllocator) = 0;
+    virtual PARTHANDLER * createFetchPartHandler(IDistributedFilePart * part, offset_t base, offset_t size, IThreadedExceptionHandler * handler, bool blockcompressed, MemoryAttr &encryptionkey, ISourceRowPrefetcher * prefetcher, IEngineRowAllocator *rowAllocator) = 0;
 };
 
 template <class PARTHANDLER, class LEFTPTR, class REQUEST>
@@ -1982,7 +1958,7 @@ class DistributedFileFetchHandler : public DistributedFileFetchHandlerBase
 public:
     typedef DistributedFileFetchHandler<PARTHANDLER, LEFTPTR, REQUEST> SELF;
 
-    DistributedFileFetchHandler(IDistributedFile * f, IFetchHandlerFactory<PARTHANDLER> & factory, MemoryAttr &encryptionkey, IOutputRowDeserializer * rowDeserializer, IEngineRowAllocator *rowAllocator) : file(f)
+    DistributedFileFetchHandler(IDistributedFile * f, IFetchHandlerFactory<PARTHANDLER> & factory, MemoryAttr &encryptionkey, ISourceRowPrefetcher * prefetcher, IEngineRowAllocator *rowAllocator) : file(f)
     {
         numParts = f->numParts();
         parts = new PARTHANDLER *[numParts];
@@ -1994,7 +1970,7 @@ public:
         {
             IDistributedFilePart *part = f->getPart(idx);
             offset_t size = getPartSize(part);
-            parts[idx] = factory.createFetchPartHandler(part, base, size, this, blockcompressed, encryptionkey, rowDeserializer, rowAllocator);
+            parts[idx] = factory.createFetchPartHandler(part, base, size, this, blockcompressed, encryptionkey, prefetcher, rowAllocator);
             base += size;
         }
         exception = NULL;
@@ -2257,7 +2233,7 @@ public:
         fetch.getFileEncryptKey(kl,k);
         MemoryAttr encryptionkey;
         encryptionkey.setOwn(kl,k);
-        parts.setown(new DistributedFileFetchHandler<SimpleFetchPartHandlerBase, const void *, FetchRequest>(f, *this, encryptionkey, rowDeserializer, rowAllocator));
+        parts.setown(new DistributedFileFetchHandler<SimpleFetchPartHandlerBase, const void *, FetchRequest>(f, *this, encryptionkey, prefetcher, rowAllocator));
     }
 
     virtual void stopParts()
@@ -2357,7 +2333,9 @@ public:
         dequeuedSeq = 0;
     }
 protected:
-    Owned<IOutputRowDeserializer> rowDeserializer;
+    Owned<ISourceRowPrefetcher> prefetcher;
+    Owned<IOutputMetaData> actualDiskMeta;
+    Owned<const IDynamicTransform> translator;
 private:
     PartHandlerThreadFactory<FetchRequest> threadFactory;   
     Owned<DistributedFileFetchHandler<SimpleFetchPartHandlerBase, const void *, FetchRequest> > parts;
@@ -2381,20 +2359,31 @@ public:
     {
         CHThorFetchActivityBase::ready();
         rowLimit = helper.getRowLimit();
-        rowDeserializer.setown(helper.queryDiskRecordSize()->createDiskDeserializer(agent.queryCodeContext(), activityId));
-        diskAllocator.setown(agent.queryCodeContext()->getRowAllocator(helper.queryDiskRecordSize(), activityId));
+    }
+
+    virtual void initParts(IDistributedFile * f) override
+    {
+        CHThorFetchActivityBase::initParts(f);
+        prefetcher.setown(actualDiskMeta->createDiskPrefetcher());
     }
 
     virtual bool needsAllocator() const { return true; }
 
     virtual void processFetch(FetchRequest const * fetch, offset_t pos, ISerialStream *rawStream)
     {
-        CThorStreamDeserializerSource deserializeSource;
-        deserializeSource.setStream(rawStream);
-        deserializeSource.reset(pos);
-        RtlDynamicRowBuilder rowBuilder(diskAllocator);
-        unsigned sizeRead = rowDeserializer->deserialize(rowBuilder.ensureRow(), deserializeSource);
-        OwnedConstRoxieRow rawBuffer(rowBuilder.finalizeRowClear(sizeRead));
+        CThorContiguousRowBuffer prefetchSource;
+        prefetchSource.setStream(rawStream);
+        prefetchSource.reset(pos);
+        prefetcher->readAhead(prefetchSource);
+        const byte *rawBuffer = prefetchSource.queryRow();
+
+        MemoryBuffer buf;
+        if (translator)
+        {
+            MemoryBufferBuilder aBuilder(buf, 0);
+            translator->translate(aBuilder, rawBuffer);
+            rawBuffer = reinterpret_cast<const byte *>(buf.toByteArray());
+        }
 
         CriticalBlock procedure(transformCrit);
         size32_t thisSize;
@@ -2422,22 +2411,51 @@ public:
         helper.onLimitExceeded();
     }
 
-    virtual SimpleFetchPartHandlerBase * createFetchPartHandler(IDistributedFilePart * part, offset_t base, offset_t size, IThreadedExceptionHandler * handler, bool blockcompressed, MemoryAttr &encryptionkey, IOutputRowDeserializer * rowDeserializer, IEngineRowAllocator *rowAllocator)
+    virtual SimpleFetchPartHandlerBase * createFetchPartHandler(IDistributedFilePart * part, offset_t base, offset_t size, IThreadedExceptionHandler * handler, bool blockcompressed, MemoryAttr &encryptionkey, ISourceRowPrefetcher * prefetcher, IEngineRowAllocator *rowAllocator)
     {
-        return new FlatFetchPartHandler(*this, part, base, size, handler, threadPool, blockcompressed, encryptionkey, activityId, outputMeta, rowDeserializer, rowAllocator);
+        return new FlatFetchPartHandler(*this, part, base, size, handler, threadPool, blockcompressed, encryptionkey, activityId, outputMeta, prefetcher, rowAllocator);
     }
 
 protected:
     virtual void verifyFetchFormatCrc(IDistributedFile * f)
     {
-        if(!agent.queryWorkUnit()->getDebugValueBool("skipFileFormatCrcCheck", false))
-            ::verifyFormatCrcSuper(helper.getDiskFormatCrc(), f, false, true);
+        actualDiskMeta.set(helper.queryDiskRecordSize());
+        translator.clear();
+        if (agent.rltEnabled()==RecordTranslationMode::None)
+        {
+            if(!agent.queryWorkUnit()->getDebugValueBool("skipFileFormatCrcCheck", false))
+                ::verifyFormatCrcSuper(helper.getDiskFormatCrc(), f, false, true);
+        }
+        else
+        {
+            bool crcMatched = ::verifyFormatCrcSuper(helper.getDiskFormatCrc(), f, false, false);  // MORE - fetch requires all to match.
+            if (!crcMatched)
+            {
+                IPropertyTree &props = f->queryAttributes();
+                actualDiskMeta.setown(getDaliLayoutInfo(props));
+                if (actualDiskMeta)
+                {
+                    translator.setown(createRecordTranslator(helper.queryProjectedDiskRecordSize()->queryRecordAccessor(true), actualDiskMeta->queryRecordAccessor(true)));
+                    if (translator->canTranslate())
+                    {
+                        if (agent.rltEnabled()==RecordTranslationMode::None)
+                            throw MakeStringException(0, "Translatable file layout mismatch reading file %s but translation disabled", f->queryLogicalName());
+#ifdef _DEBUG
+                        translator->describe();
+#endif
+                    }
+                    else
+                        throw MakeStringException(0, "Untranslatable file layout mismatch reading file %s", f->queryLogicalName());
+                }
+                else
+                    throw MakeStringException(0, "Untranslatable file layout mismatch reading file %s - key layout information not found", f->queryLogicalName());
+            }
+        }
     }
 
 protected:
     CriticalSection transformCrit;
     IHThorFetchArg & helper;
-    Owned<IEngineRowAllocator> diskAllocator;
 };
 
 extern HTHOR_API IHThorActivity *createFetchActivity(IAgentContext &_agent, unsigned _activityId, unsigned _subgraphId, IHThorFetchArg &arg, ThorActivityKind _kind)
@@ -2527,7 +2545,6 @@ public:
     {
         CHThorFetchActivityBase::ready();
         rowLimit = helper.getRowLimit();
-        rowDeserializer.setown(helper.queryDiskRecordSize()->createDiskDeserializer(agent.queryCodeContext(), activityId));
     }
 
     virtual void onLimitExceeded()
@@ -2535,9 +2552,9 @@ public:
         helper.onLimitExceeded();
     }
 
-    virtual SimpleFetchPartHandlerBase * createFetchPartHandler(IDistributedFilePart * part, offset_t base, offset_t size, IThreadedExceptionHandler * handler, bool blockcompressed, MemoryAttr &encryptionkey, IOutputRowDeserializer * rowDeserializer, IEngineRowAllocator *rowAllocator)
+    virtual SimpleFetchPartHandlerBase * createFetchPartHandler(IDistributedFilePart * part, offset_t base, offset_t size, IThreadedExceptionHandler * handler, bool blockcompressed, MemoryAttr &encryptionkey, ISourceRowPrefetcher * prefetcher, IEngineRowAllocator *rowAllocator)
     {
-        return new FlatFetchPartHandler(*this, part, base, size, handler, threadPool, blockcompressed, encryptionkey, activityId, outputMeta, rowDeserializer, rowAllocator);
+        return new FlatFetchPartHandler(*this, part, base, size, handler, threadPool, blockcompressed, encryptionkey, activityId, outputMeta, prefetcher, rowAllocator);
     }
 
 protected:
@@ -2675,7 +2692,7 @@ public:
         helper.onLimitExceeded();
     }
 
-    virtual SimpleFetchPartHandlerBase * createFetchPartHandler(IDistributedFilePart * part, offset_t base, offset_t size, IThreadedExceptionHandler * handler, bool blockcompressed, MemoryAttr &encryptionkey, IOutputRowDeserializer * rowDeserializer, IEngineRowAllocator *rowAllocator)
+    virtual SimpleFetchPartHandlerBase * createFetchPartHandler(IDistributedFilePart * part, offset_t base, offset_t size, IThreadedExceptionHandler * handler, bool blockcompressed, MemoryAttr &encryptionkey, ISourceRowPrefetcher * prefetcher, IEngineRowAllocator *rowAllocator)
     {
         return new XmlFetchPartHandler(*this, part, base, size, handler, 4096, threadPool, blockcompressed, encryptionkey, activityId, outputMeta, kind==TAKjsonfetch); //MORE: need to put correct stream buffer size here, when Gavin provides it
     }
@@ -3285,8 +3302,8 @@ public:
 class KeyedJoinFetchPartHandler : public FetchPartHandlerBase, public ThreadedPartHandler<KeyedJoinFetchRequest>
 {
 public:
-    KeyedJoinFetchPartHandler(IKeyedJoinFetchHandlerCallback & _owner, IDistributedFilePart *_part, offset_t _base, offset_t _size, IThreadedExceptionHandler *_handler, IThreadPool * _threadPool, bool _blockcompressed, MemoryAttr &_encryptionkey, unsigned _activityId, CachedOutputMetaData const & _outputMeta, IOutputRowDeserializer * _rowDeserializer, IEngineRowAllocator *_rowAllocator)
-        : FetchPartHandlerBase(_base, _size, _blockcompressed, _encryptionkey, _activityId, _outputMeta, _rowDeserializer, _rowAllocator),
+    KeyedJoinFetchPartHandler(IKeyedJoinFetchHandlerCallback & _owner, IDistributedFilePart *_part, offset_t _base, offset_t _size, IThreadedExceptionHandler *_handler, IThreadPool * _threadPool, bool _blockcompressed, MemoryAttr &_encryptionkey, unsigned _activityId, CachedOutputMetaData const & _outputMeta, ISourceRowPrefetcher * _prefetcher, IEngineRowAllocator *_rowAllocator)
+        : FetchPartHandlerBase(_base, _size, _blockcompressed, _encryptionkey, _activityId, _outputMeta, _prefetcher, _rowAllocator),
           ThreadedPartHandler<KeyedJoinFetchRequest>(_part, _handler, _threadPool),
           owner(_owner)
     {
@@ -3353,10 +3370,10 @@ class CHThorKeyedJoinActivity  : public CHThorThreadedActivityBase, implements I
     IDistributedFile * dFile;
     IDistributedSuperFile * super;
     CachedOutputMetaData eclKeySize;
-    Owned<IOutputRowDeserializer> rowDeserializer;
-    Owned<IEngineRowAllocator> diskAllocator;
-    IPointerArrayOf<IOutputMetaData> actualLayouts;
-
+    Owned<ISourceRowPrefetcher> prefetcher;
+    IPointerArrayOf<IOutputMetaData> actualLayouts;  // all the index layouts are saved in here to ensure their lifetime is adequate
+    Owned<IOutputMetaData> actualDiskMeta;           // only one disk layout is permitted
+    Owned<const IDynamicTransform> translator;
 public:
     CHThorKeyedJoinActivity(IAgentContext &_agent, unsigned _activityId, unsigned _subgraphId, IHThorKeyedJoinArg &_arg, ThorActivityKind _kind)
         : CHThorThreadedActivityBase(_agent, _activityId, _subgraphId, _arg, _arg, _kind, _arg.queryDiskRecordSize()), helper(_arg)
@@ -3406,11 +3423,6 @@ public:
                 defaultRight.setown(rowBuilder.finalizeRowClear(thisSize));
             }
         }
-        if (needsDiskRead)
-        {
-            rowDeserializer.setown(helper.queryDiskRecordSize()->createDiskDeserializer(agent.queryCodeContext(), activityId));
-            diskAllocator.setown(agent.queryCodeContext()->getRowAllocator(helper.queryDiskRecordSize(), activityId));
-        }
     }
 
     virtual void stop()
@@ -3435,7 +3447,8 @@ public:
         if (needsDiskRead)
         {
             inputRowAllocator.setown(agent.queryCodeContext()->getRowAllocator(helper.queryDiskRecordSize(), activityId));
-            parts.setown(new DistributedFileFetchHandler<KeyedJoinFetchPartHandler, MatchSet *, KeyedJoinFetchRequest>(f, *this, encryptionkey, rowDeserializer, inputRowAllocator));
+            parts.setown(new DistributedFileFetchHandler<KeyedJoinFetchPartHandler, MatchSet *, KeyedJoinFetchRequest>(f, *this, encryptionkey, prefetcher, inputRowAllocator));
+            prefetcher.setown(actualDiskMeta->createDiskPrefetcher());
         }
     }
 
@@ -3527,20 +3540,26 @@ public:
         stopThread();
     }
 
-    virtual KeyedJoinFetchPartHandler * createFetchPartHandler(IDistributedFilePart * part, offset_t base, offset_t size, IThreadedExceptionHandler * handler, bool blockcompressed, MemoryAttr &encryptionkey, IOutputRowDeserializer * rowDeserializer, IEngineRowAllocator *rowAllocator)
+    virtual KeyedJoinFetchPartHandler * createFetchPartHandler(IDistributedFilePart * part, offset_t base, offset_t size, IThreadedExceptionHandler * handler, bool blockcompressed, MemoryAttr &encryptionkey, ISourceRowPrefetcher * prefetcher, IEngineRowAllocator *rowAllocator)
     {
-        return new KeyedJoinFetchPartHandler(*this, part, base, size, handler, threadPool, blockcompressed, encryptionkey, activityId, outputMeta, rowDeserializer, rowAllocator);
+        return new KeyedJoinFetchPartHandler(*this, part, base, size, handler, threadPool, blockcompressed, encryptionkey, activityId, outputMeta, prefetcher, rowAllocator);
     }
 
     virtual void processFetch(KeyedJoinFetchRequest const * fetch, offset_t pos, ISerialStream *rawStream)
     {
-        CThorStreamDeserializerSource deserializeSource;
-        deserializeSource.setStream(rawStream);
-        deserializeSource.reset(pos);
-        RtlDynamicRowBuilder rowBuilder(diskAllocator);
-        unsigned sizeRead = rowDeserializer->deserialize(rowBuilder.ensureRow(), deserializeSource);
-        OwnedConstRoxieRow row = rowBuilder.finalizeRowClear(sizeRead);
+        CThorContiguousRowBuffer prefetchSource;
+        prefetchSource.setStream(rawStream);
+        prefetchSource.reset(pos);
+        prefetcher->readAhead(prefetchSource);
+        const byte *row = prefetchSource.queryRow();
 
+        MemoryBuffer buf;
+        if (translator)
+        {
+            MemoryBufferBuilder aBuilder(buf, 0);
+            translator->translate(aBuilder, row);
+            row = reinterpret_cast<const byte *>(buf.toByteArray());
+        }
         if(match(fetch->ms, row))
         {
             if(exclude)
@@ -4019,7 +4038,7 @@ protected:
             return NULL;
         }
 
-        if(!rltEnabled(agent.queryWorkUnit()))
+        if(agent.rltEnabled() == RecordTranslationMode::None)
         {
             verifyFormatCrc(helper.getIndexFormatCrc(), f, super ? super->queryLogicalName() : NULL, true, true);
             return NULL;
@@ -4030,32 +4049,17 @@ protected:
             return NULL;
         }
 
-        Owned<IOutputMetaData> actualFormat;
         IPropertyTree &props = f->queryAttributes();
-        bool isGrouped = props.getPropBool("@grouped", false);
-        if (props.hasProp("_rtlType"))
-        {
-            MemoryBuffer layoutBin;
-            props.getPropBin("_rtlType", layoutBin);
-            actualFormat.setown(createTypeInfoOutputMetaData(layoutBin, isGrouped, nullptr));
-        }
-        else if (props.hasProp("ECL"))
-        {
-            StringBuffer layoutECL;
-            props.getProp("ECL", layoutECL);
-            MultiErrorReceiver errs;
-            Owned<IHqlExpression> expr = parseQuery(layoutECL.str(), &errs);
-            if (errs.errCount() == 0)
-            {
-                MemoryBuffer layoutBin;
-                if (exportBinaryType(layoutBin, expr))
-                    actualFormat.setown(createTypeInfoOutputMetaData(layoutBin, isGrouped, nullptr));
-            }
-        }
+        Owned<IOutputMetaData> actualFormat = getDaliLayoutInfo(props);
         if (actualFormat)
         {
             actualLayouts.append(actualFormat.getLink());  // ensure adequate lifespan
-            return createRecordTranslator(helper.queryProjectedIndexRecordSize()->queryRecordAccessor(true), actualFormat->queryRecordAccessor(true));
+            Owned<const IDynamicTransform> payloadTranslator =  createRecordTranslator(helper.queryProjectedIndexRecordSize()->queryRecordAccessor(true), actualFormat->queryRecordAccessor(true));
+            if (!payloadTranslator->canTranslate())
+                throw MakeStringException(0, "Untranslatable key layout mismatch reading index %s", f->queryLogicalName());
+            if (payloadTranslator->keyedTranslated())
+                throw MakeStringException(0, "Untranslatable key layout mismatch reading index %s - keyed fields do not match", f->queryLogicalName());
+            return payloadTranslator.getClear();
         }
         throw MakeStringException(0, "Untranslatable key layout mismatch reading index %s - key layout information not found", f->queryLogicalName());
     }
@@ -4080,8 +4084,35 @@ protected:
 
     virtual void verifyFetchFormatCrc(IDistributedFile * f)
     {
-        if(!agent.queryWorkUnit()->getDebugValueBool("skipFileFormatCrcCheck", false))
-            ::verifyFormatCrcSuper(helper.getDiskFormatCrc(), f, false, true);
+        actualDiskMeta.set(helper.queryDiskRecordSize());
+        translator.clear();
+        if (agent.rltEnabled()==RecordTranslationMode::None)
+        {
+            if(!agent.queryWorkUnit()->getDebugValueBool("skipFileFormatCrcCheck", false))
+                ::verifyFormatCrcSuper(helper.getDiskFormatCrc(), f, false, true);
+        }
+        else
+        {
+            bool crcMatched = ::verifyFormatCrcSuper(helper.getDiskFormatCrc(), f, false, false);  // MORE - fetch requires all to match.
+            if (!crcMatched)
+            {
+                IPropertyTree &props = f->queryAttributes();
+                actualDiskMeta.setown(getDaliLayoutInfo(props));
+                if (actualDiskMeta)
+                {
+                    translator.setown(createRecordTranslator(helper.queryProjectedDiskRecordSize()->queryRecordAccessor(true), actualDiskMeta->queryRecordAccessor(true)));
+                    if (translator->canTranslate())
+                    {
+                        if (agent.rltEnabled()==RecordTranslationMode::None)
+                            throw MakeStringException(0, "Translatable file layout mismatch reading file %s but translation disabled", f->queryLogicalName());
+                    }
+                    else
+                        throw MakeStringException(0, "Untranslatable file layout mismatch reading file %s", f->queryLogicalName());
+                }
+                else
+                    throw MakeStringException(0, "Untranslatable file layout mismatch reading file %s - key layout information not found", f->queryLogicalName());
+            }
+        }
     }
 
     virtual const RtlRecord &queryIndexRecord()

roxie/ccd/CMakeLists.txt

@@ -70,7 +70,6 @@ include_directories (
          ${HPCC_SOURCE_DIR}/roxie/udplib
          ${HPCC_SOURCE_DIR}/roxie/roxie
          ${HPCC_SOURCE_DIR}/common/environment
-         ${HPCC_SOURCE_DIR}/ecl/hql
          ${HPCC_SOURCE_DIR}/ecl/hthor
          ${HPCC_SOURCE_DIR}/ecl/schedulectrl
          ${HPCC_SOURCE_DIR}/rtl/nbcd
@@ -105,7 +104,6 @@ install ( TARGETS ccd RUNTIME DESTINATION ${EXEC_DIR} LIBRARY DESTINATION ${LIB_
 target_link_libraries ( ccd 
          jlib
          nbcd
-         hql
          roxiemem 
          udplib 
          remote 

roxie/ccd/ccdfile.cpp

@@ -41,9 +41,9 @@
 #include <sys/syscall.h>
 #include "ioprio.h"
 #endif
+#include "thorcommon.hpp"
 #include "eclhelper_dyn.hpp"
 #include "rtldynfield.hpp"
-#include "hqlexpr.hpp"
 
 atomic_t numFilesOpen[2];
 
@@ -1831,25 +1831,8 @@ protected:
         if (formatCrcs.length() && formatCrc == formatCrcs.item(prevIdx) &&
             diskTypeInfo.item(prevIdx) && isGrouped==diskTypeInfo.item(prevIdx)->isGrouped())
             actualFormat.set(diskTypeInfo.item(prevIdx));
-        else if (props.hasProp("_rtlType"))
-        {
-            MemoryBuffer layoutBin;
-            props.getPropBin("_rtlType", layoutBin);
-            actualFormat.setown(createTypeInfoOutputMetaData(layoutBin, isGrouped, nullptr));
-        }
-        else if (props.hasProp("ECL"))
-        {
-            StringBuffer layoutECL;
-            props.getProp("ECL", layoutECL);
-            MultiErrorReceiver errs;
-            Owned<IHqlExpression> expr = parseQuery(layoutECL.str(), &errs);
-            if (errs.errCount() == 0)
-            {
-                MemoryBuffer layoutBin;
-                if (exportBinaryType(layoutBin, expr))
-                    actualFormat.setown(createTypeInfoOutputMetaData(layoutBin, isGrouped, nullptr));
-            }
-        }
+        else
+            actualFormat.setown(getDaliLayoutInfo(props));
         diskTypeInfo.append(actualFormat.getClear());
         formatCrcs.append(formatCrc);
 

roxie/ccd/ccdmain.cpp

@@ -837,14 +837,7 @@ int STARTQUERY_API start_query(int argc, const char *argv[])
         fieldTranslationEnabled = RecordTranslationMode::None;
         const char *val = topology->queryProp("@fieldTranslationEnabled");
         if (val)
-        {
-            if (strieq(val, "alwaysDisk"))
-                fieldTranslationEnabled = RecordTranslationMode::AlwaysDisk;
-            else if (strieq(val, "alwaysECL"))
-                fieldTranslationEnabled = RecordTranslationMode::AlwaysECL;
-            else if (strieq(val, "payload") || strToBool(val))
-                fieldTranslationEnabled = RecordTranslationMode::Payload;
-        }
+            fieldTranslationEnabled = getTranslationMode(val);
 
         pretendAllOpt = topology->getPropBool("@ignoreMissingFiles", false);
         memoryStatsInterval = topology->getPropInt("@memoryStatsInterval", 60);

roxie/ccd/ccdquery.cpp

@@ -401,16 +401,7 @@ void QueryOptions::updateFromWorkUnit(RecordTranslationMode &value, IConstWorkUn
     SCMStringBuffer val;
     wu.getDebugValue(name, val);
     if (val.length())
-    {
-        if (strieq(val.str(), "alwaysDisk"))
-            value = RecordTranslationMode::AlwaysDisk;
-        else if (strieq(val.str(), "alwaysECL"))
-            value = RecordTranslationMode::AlwaysECL;
-        else if (strieq(val.str(), "payload") || strToBool(val.str()))
-            value = RecordTranslationMode::Payload;
-        else
-            value = RecordTranslationMode::None;
-    }
+        value = getTranslationMode(val.str());
 }
 
 void QueryOptions::setFromContext(const IPropertyTree *ctx)

roxie/ccd/ccdstate.cpp

@@ -2211,24 +2211,10 @@ private:
             {
                 const char *val = control->queryProp("@val");
                 if (val)
-                {
-                    if (strieq(val, "alwaysDisk"))
-                        fieldTranslationEnabled = RecordTranslationMode::AlwaysDisk;
-                    else if (strieq(val, "alwaysECL"))
-                        fieldTranslationEnabled = RecordTranslationMode::AlwaysECL;
-                    else if (!val || strToBool(val) || strieq(val, "payload"))
-                    {
-                        fieldTranslationEnabled = RecordTranslationMode::Payload;
-                        val = "payload";
-                    }
-                    else
-                    {
-                        fieldTranslationEnabled = RecordTranslationMode::None;
-                        val = "false";
-                    }
-                }
+                    fieldTranslationEnabled = getTranslationMode(val);
                 else
-                    val = "false";
+                    fieldTranslationEnabled = RecordTranslationMode::Payload;
+                val = getTranslationModeText(fieldTranslationEnabled);
                 topology->setProp("@fieldTranslationEnabled", val);
             }
             else if (stricmp(queryName, "control:flushJHtreeCacheOnOOM")==0)

rtl/eclrtl/rtldynfield.cpp

@@ -34,6 +34,33 @@
 
 //---------------------------------------------------------------------------------------------------------------------
 
+extern ECLRTL_API RecordTranslationMode getTranslationMode(const char *val)
+{
+    if (strieq(val, "alwaysDisk"))
+        return RecordTranslationMode::AlwaysDisk;
+    else if (strieq(val, "alwaysECL"))
+        return RecordTranslationMode::AlwaysECL;
+    else if (!val || strToBool(val) || strieq(val, "payload"))
+        return RecordTranslationMode::Payload;
+    else
+        return RecordTranslationMode::None;
+}
+
+extern ECLRTL_API const char *getTranslationModeText(RecordTranslationMode val)
+{
+    switch (val)
+    {
+    case RecordTranslationMode::AlwaysDisk: return "alwaysDisk";
+    case RecordTranslationMode::AlwaysECL: return "alwaysECL";
+    case RecordTranslationMode::Payload: return "payload";
+    case RecordTranslationMode::None: return "off";
+    }
+    throwUnexpected();
+}
+
+
+//---------------------------------------------------------------------------------------------------------------------
+
 const RtlTypeInfo *FieldTypeInfoStruct::createRtlTypeInfo(IThorIndexCallback *_callback) const
 {
     const RtlTypeInfo *ret = nullptr;
@@ -1643,7 +1670,7 @@ public:
             }
         }
     }
-    virtual bool translate(RowFilter &filters) const override
+    virtual bool translate(RowFilter &filters) const override // MORE - remove this version - it's not safe
     {
         bool mapNeeded = false;
         if (translateNeeded)
@@ -1671,6 +1698,34 @@ public:
         }
         return mapNeeded;
     }
+    virtual bool translate(RowFilter &filter, IConstArrayOf<IFieldFilter> &in) const override
+    {
+        bool mapNeeded = false;
+        if (translateNeeded)
+        {
+            unsigned numFields = in.length();
+            for (unsigned idx = 0; idx < numFields; idx++)
+            {
+                unsigned fieldNum = in.item(idx).queryFieldIndex();
+                unsigned mappedFieldNum = map.isItem(fieldNum) ? map.item(fieldNum) : (unsigned) -1;
+                if (mappedFieldNum != fieldNum)
+                {
+                    mapNeeded = true;
+                    switch (mappedFieldNum)
+                    {
+                    case (unsigned) -1: throw makeStringExceptionV(0, "Cannot translate keyed filter on field %u - no matching field", idx);
+                    case (unsigned) -2: throw makeStringExceptionV(0, "Cannot translate keyed filter on field %u - incompatible matching field type", idx);
+                    default:
+                        filter.addFilter(*in.item(idx).remap(mappedFieldNum));
+                        break;
+                    }
+                }
+                else
+                    filter.addFilter(OLINK(in.item(idx)));
+            }
+        }
+        return mapNeeded;
+    }
     virtual bool needsTranslate() const
     {
         return translateNeeded;

rtl/eclrtl/rtldynfield.hpp

@@ -19,6 +19,7 @@
 #define rtldynfield_hpp
 
 #include "rtlfield.hpp"
+#include "rtlnewkey.hpp"
 
 //These classes support the dynamic creation of type and field information
 
@@ -104,6 +105,9 @@ interface IRtlFieldTypeDeserializer : public IInterface
 
 enum class RecordTranslationMode { None = 0, All = 1, Payload = 2, AlwaysDisk = 3, AlwaysECL = 4 };  // Latter 2 are for testing purposes only
 
+extern ECLRTL_API RecordTranslationMode getTranslationMode(const char *modeStr);
+extern ECLRTL_API const char *getTranslationModeText(RecordTranslationMode val);
+
 interface IDynamicTransform : public IInterface
 {
     virtual void describe() const = 0;
@@ -114,11 +118,11 @@ interface IDynamicTransform : public IInterface
     virtual bool keyedTranslated() const = 0;
 };
 
-class RowFilter;
 interface IKeyTranslator : public IInterface
 {
     virtual void describe() const = 0;
-    virtual bool translate(RowFilter &filters) const = 0;
+    virtual bool translate(RowFilter &filters) const = 0;  // MORE - this should be deprecated
+    virtual bool translate(RowFilter &result, IConstArrayOf<IFieldFilter> &_filters) const = 0;
     virtual bool needsTranslate() const = 0;
 };
 

rtl/eclrtl/rtlnewkey.cpp

@@ -1343,6 +1343,14 @@ bool RowFilter::matches(const RtlRow & row) const
     return true;
 }
 
+void RowFilter::appendFilters(IConstArrayOf<IFieldFilter> & _filters)
+{
+    ForEachItemIn(i, _filters)
+    {
+        addFilter(OLINK(_filters.item(i)));
+    }
+}
+
 void RowFilter::extractKeyFilter(const RtlRecord & record, IConstArrayOf<IFieldFilter> & keyFilters) const
 {
     if (!filters)
@@ -1424,6 +1432,12 @@ void RowFilter::remove(unsigned idx)
     filters.remove(idx);
 }
 
+void RowFilter::clear()
+{
+    filters.kill();
+    numFieldsRequired = 0;
+}
+
 void RowFilter::recalcFieldsRequired()
 {
     numFieldsRequired = 0;

rtl/eclrtl/rtlnewkey.hpp

@@ -44,6 +44,8 @@ public:
     void remapField(unsigned filterIdx, unsigned newFieldNum);
     void recalcFieldsRequired();
     void remove(unsigned idx);
+    void clear();
+    void appendFilters(IConstArrayOf<IFieldFilter> &_filters);
 protected:
     IConstArrayOf<IFieldFilter> filters;
     unsigned numFieldsRequired = 0;

testing/regress/ecl/translatedisk.ecl

@@ -16,7 +16,6 @@
 ############################################################################## */
 
 //nothor
-//nohthor
 //version multiPart=false
 //version multiPart=true