Merge pull request #9307 from jakesmith/hpcc-16582

HPCC-16582 Ensure logical filename issued on index read error

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

thorlcr/activities/fetch/thfetchslave.cpp

@@ -81,6 +81,7 @@ protected:
     Owned<IRowStream> keyOutStream;
     CActivityBase &owner;
     Linked<IThorRowInterfaces> keyRowIf, fetchRowIf;
+    StringAttr logicalFilename;
 
     class CFPosHandler : implements IHash, public CSimpleInterface
     {
@@ -122,8 +123,9 @@ protected:
 public:
     IMPLEMENT_IINTERFACE_USING(CSimpleInterface);
 
-    CFetchStream(CActivityBase &_owner, IThorRowInterfaces *_keyRowIf, IThorRowInterfaces *_fetchRowIf, bool &_abortSoon, CPartDescriptorArray &_parts, unsigned _offsetCount, size32_t offsetMapSz, const void *offsetMap, IFetchHandler *_iFetchHandler, mptag_t _tag, IExpander *_eexp)
-        : owner(_owner), keyRowIf(_keyRowIf), fetchRowIf(_fetchRowIf), abortSoon(_abortSoon), iFetchHandler(_iFetchHandler), offsetCount(_offsetCount), tag(_tag), eexp(_eexp)
+    CFetchStream(CActivityBase &_owner, IThorRowInterfaces *_keyRowIf, IThorRowInterfaces *_fetchRowIf, bool &_abortSoon, const char *_logicalFilename, CPartDescriptorArray &_parts, unsigned _offsetCount, size32_t offsetMapSz, const void *offsetMap, IFetchHandler *_iFetchHandler, mptag_t _tag, IExpander *_eexp)
+        : owner(_owner), keyRowIf(_keyRowIf), fetchRowIf(_fetchRowIf), abortSoon(_abortSoon), logicalFilename(_logicalFilename),
+          iFetchHandler(_iFetchHandler), offsetCount(_offsetCount), tag(_tag), eexp(_eexp)
     {
         distributor = NULL;
         fposHash = NULL;
@@ -155,7 +157,7 @@ public:
                 e->top = e->base + part.queryProperties().getPropInt64("@size");
                 e->index = f;
 
-                Owned<IDelayedFile> lfile = queryThor().queryFileCache().lookup(owner, part);
+                Owned<IDelayedFile> lfile = queryThor().queryFileCache().lookup(owner, logicalFilename, part);
                 e->file = lfile.getClear();
             }
         }
@@ -262,9 +264,9 @@ public:
 };
 
 
-IFetchStream *createFetchStream(CSlaveActivity &owner, IThorRowInterfaces *keyRowIf, IThorRowInterfaces *fetchRowIf, bool &abortSoon, CPartDescriptorArray &parts, unsigned offsetCount, size32_t offsetMapSz, const void *offsetMap, IFetchHandler *iFetchHandler, mptag_t tag, IExpander *eexp)
+IFetchStream *createFetchStream(CSlaveActivity &owner, IThorRowInterfaces *keyRowIf, IThorRowInterfaces *fetchRowIf, bool &abortSoon, const char *logicalFilename, CPartDescriptorArray &parts, unsigned offsetCount, size32_t offsetMapSz, const void *offsetMap, IFetchHandler *iFetchHandler, mptag_t tag, IExpander *eexp)
 {
-    return new CFetchStream(owner, keyRowIf, fetchRowIf, abortSoon, parts, offsetCount, offsetMapSz, offsetMap, iFetchHandler, tag, eexp);
+    return new CFetchStream(owner, keyRowIf, fetchRowIf, abortSoon, logicalFilename, parts, offsetCount, offsetMapSz, offsetMap, iFetchHandler, tag, eexp);
 }
 
 class CFetchSlaveBase : public CSlaveActivity, implements IFetchHandler
@@ -456,7 +458,7 @@ public:
         }
 
         Owned<IThorRowInterfaces> rowIf = createRowInterfaces(queryRowMetaData());
-        fetchStream = createFetchStream(*this, keyInIf, rowIf, abortSoon, parts, offsetCount, offsetMapSz, offsetMapBytes.toByteArray(), this, mptag, eexp);
+        fetchStream = createFetchStream(*this, keyInIf, rowIf, abortSoon, fetchContext->getFileName(), parts, offsetCount, offsetMapSz, offsetMapBytes.toByteArray(), this, mptag, eexp);
         fetchStreamOut = fetchStream->queryOutput();
         fetchStream->start(keyIn);
         initializeFileParts();

thorlcr/activities/fetch/thfetchslave.ipp

@@ -39,7 +39,7 @@ interface IFetchStream : extends IInterface
     virtual void abort() = 0;
 };
 
-IFetchStream *createFetchStream(CSlaveActivity &owner, IThorRowInterfaces *keyRowIf, IThorRowInterfaces *fetchRowIf, bool &abortSoon, CPartDescriptorArray &parts, unsigned offsetCount, size32_t offsetMapSz, const void *offsetMap, IFetchHandler *iFetchHandler, mptag_t tag, IExpander *eexp=NULL);
+IFetchStream *createFetchStream(CSlaveActivity &owner, IThorRowInterfaces *keyRowIf, IThorRowInterfaces *fetchRowIf, bool &abortSoon, const char *logicalFilename, CPartDescriptorArray &parts, unsigned offsetCount, size32_t offsetMapSz, const void *offsetMap, IFetchHandler *iFetchHandler, mptag_t tag, IExpander *eexp=NULL);
 
 activityslaves_decl CActivityBase *createFetchSlave(CGraphElementBase *container);
 activityslaves_decl CActivityBase *createCsvFetchSlave(CGraphElementBase *container);

thorlcr/activities/indexread/thindexreadslave.cpp

@@ -138,7 +138,7 @@ protected:
         rfn.getPath(filePath);
         unsigned crc=0;
         partDesc.getCrc(crc);
-        Owned<IDelayedFile> lfile = queryThor().queryFileCache().lookup(*this, partDesc);
+        Owned<IDelayedFile> lfile = queryThor().queryFileCache().lookup(*this, logicalFilename, partDesc);
         return createKeyIndex(filePath.str(), crc, *lfile, false, false);
     }
     const void *nextKey()

thorlcr/activities/keyedjoin/thkeyedjoinslave.cpp

@@ -1882,7 +1882,7 @@ public:
                 filePart.getFilename(0, rfn);
                 StringBuffer filename;
                 rfn.getPath(filename);
-                Owned<IDelayedFile> lfile = queryThor().queryFileCache().lookup(*this, filePart);
+                Owned<IDelayedFile> lfile = queryThor().queryFileCache().lookup(*this, indexName, filePart);
                 partKeySet->addIndex(createKeyIndex(filename.str(), crc, *lfile, false, false));
             }
             while (ip<numIndexParts);
@@ -1982,7 +1982,7 @@ public:
                 unsigned i=0;
                 for(; i<dataParts.ordinality(); i++)
                 {
-                    Owned<IDelayedFile> dFile = queryThor().queryFileCache().lookup(*this, dataParts.item(i), eexp);
+                    Owned<IDelayedFile> dFile = queryThor().queryFileCache().lookup(*this, indexName, dataParts.item(i), eexp);
                     fetchFiles.append(*dFile.getClear());
                 }
             }

thorlcr/graph/thgraph.hpp

@@ -1137,7 +1137,7 @@ interface IExpander;
 interface IThorFileCache : extends IInterface
 {
     virtual bool remove(IDelayedFile &dFile) = 0;
-    virtual IDelayedFile *lookup(CActivityBase &activity, IPartDescriptor &partDesc, IExpander *expander=NULL) = 0;
+    virtual IDelayedFile *lookup(CActivityBase &activity, const char *logicalFilenae, IPartDescriptor &partDesc, IExpander *expander=NULL) = 0;
 };
 
 class graph_decl CThorResourceBase : implements IThorResource, public CInterface

thorlcr/graph/thgraphslave.cpp

@@ -1933,7 +1933,7 @@ public:
         CriticalBlock b(crit);
         return _remove(*lFile);
     }
-    virtual IDelayedFile *lookup(CActivityBase &activity, IPartDescriptor &partDesc, IExpander *expander)
+    virtual IDelayedFile *lookup(CActivityBase &activity, const char *logicalFilename, IPartDescriptor &partDesc, IExpander *expander)
     {
         StringBuffer filename;
         RemoteFilename rfn;
@@ -1943,7 +1943,7 @@ public:
         Linked<CLazyFileIO> file = files.find(filename.str());
         if (!file)
         {
-            Owned<IReplicatedFile> repFile = createEnsurePrimaryPartFile(activity, filename.str(), &partDesc);
+            Owned<IReplicatedFile> repFile = createEnsurePrimaryPartFile(activity, logicalFilename, &partDesc);
             bool compressed = partDesc.queryOwner().isCompressed();
             file.setown(new CLazyFileIO(*this, filename.str(), repFile.getClear(), compressed, expander));
         }