HPCC-14417 Further JSON parsing and fileposition fixes

Make JSON pull parser interface getNext() compatible with XML pull parser.

Fix roxie filepostion support on XML record handler.

Signed-off-by: Anthony Fishbeck <anthony.fishbeck@lexisnexis.com>

ecl/hthor/hthorkey.cpp

@@ -2598,7 +2598,7 @@ public:
                 part->getFilename(rfn).getPath(fname);
                 throw owner.makeWrappedException(e, fname.str());
             }
-            if(!gotNext && !lastMatch) //unfortunately json parser next() has slightly different behavior, tricky, may fix later
+            if(!gotNext)
             {
                 StringBuffer fname;
                 RemoteFilename rfn;

roxie/ccd/ccdactivities.cpp

@@ -1608,19 +1608,34 @@ public:
 
 // RecordProcessor used by XML read activity. We don't try to index these or optimize fixed size cases...
 
-class XmlRecordProcessor : public RecordProcessor, implements IXMLSelect
+class XmlRecordProcessor : public RecordProcessor, implements IXMLSelect, implements IThorDiskCallback
 {
 public:
     IMPLEMENT_IINTERFACE;
     XmlRecordProcessor(CRoxieXmlReadActivity &_owner, IDirectReader *_reader)
-        : RecordProcessor(NULL), owner(_owner), reader(_reader)
+        : RecordProcessor(NULL), owner(_owner), reader(_reader), fileposition(0)
     {
         helper = _owner.helper;
-        helper->setCallback(reader->queryThorDiskCallback());
+        helper->setCallback(this);
+    }
+
+    //interface IThorDiskCallback
+    virtual unsigned __int64 getFilePosition(const void * row)
+    {
+        return fileposition;
+    }
+    virtual unsigned __int64 getLocalFilePosition(const void * row)
+    {
+        return reader->makeFilePositionLocal(fileposition);
+    }
+    virtual const char * queryLogicalFilename(const void * row)
+    {
+        return reader->queryThorDiskCallback()->queryLogicalFilename(row);
     }
 
     virtual void match(IColumnProvider &entry, offset_t startOffset, offset_t endOffset)
     {
+        fileposition = startOffset;
         lastMatch.set(&entry);
     }
 
@@ -1646,7 +1661,7 @@ public:
                 break;
             else if (lastMatch)
             {
-                unsigned transformedSize = owner.doTransform(output, rowTransformer, lastMatch, reader->queryThorDiskCallback());
+                unsigned transformedSize = owner.doTransform(output, rowTransformer, lastMatch, this);
                 lastMatch.clear();
                 if (transformedSize)
                 {
@@ -1687,6 +1702,7 @@ protected:
 
     Owned<IColumnProvider> lastMatch;
     Owned<IDirectReader> reader;
+    unsigned __int64 fileposition;
 };
 
 IInMemoryFileProcessor *createCsvRecordProcessor(CRoxieCsvReadActivity &owner, IDirectReader *_reader, bool _skipHeader, const IResolvedFile *datafile, size32_t maxRowSize)
@@ -4499,7 +4515,7 @@ public:
         try
         {
             while(!lastMatch)
-            if(!parser->next() && !lastMatch) //unfortunately json parser next() has slightly different behavior, tricky, may fix later
+            if(!parser->next())
                 throw MakeStringException(ROXIE_RECORD_FETCH_ERROR, "XML parse error at position %" I64F "d", pos);
             IHThorXmlFetchArg *h = (IHThorXmlFetchArg *) helper;
             unsigned thisSize = h->transform(rowBuilder, lastMatch, inputData, rawpos);

roxie/ccd/ccdkey.cpp

@@ -71,6 +71,26 @@ class PtrToOffsetMapper
         return fragments[a-1];
     }
 
+    FragmentInfo &findBase(offset_t pos) const
+    {
+        // MORE - could cache last hit
+        unsigned int a = 0;
+        int b = numBases;
+        int rc;
+        while ((int)a<b)
+        {
+            int i = (a+b+1)/2;
+            rc = pos - fragments[i-1].baseOffset;
+            if (rc>=0)
+                a = i;
+            else
+                b = i-1;
+        }
+        assertex(a > 0);    // or incoming value was not part of ANY fragment
+        // MORE - could also check it's in the range for this fragment
+        return fragments[a-1];
+    }
+
     PtrToOffsetMapper(const PtrToOffsetMapper &);  // Not implemented
 
 public:
@@ -132,6 +152,12 @@ public:
         return makeLocalFposOffset(frag.partNo, ptr - frag.base);
     }
 
+    offset_t makeFilePositionLocal(offset_t pos)
+    {
+        FragmentInfo &frag = findBase(pos);
+        return makeLocalFposOffset(frag.partNo, pos - frag.baseOffset);
+    }
+
     void splitPos(offset_t &offset, offset_t &size, unsigned partNo, unsigned numParts)
     {
         assert(numParts > 0);
@@ -638,6 +664,11 @@ public:
         return baseMap.ptrToLocalFilePosition(_ptr);
     }
 
+    virtual unsigned __int64 makeFilePositionLocal(offset_t pos)
+    {
+        return baseMap.makeFilePositionLocal(pos);
+    }
+
     virtual const char * queryLogicalFilename(const void * row) 
     { 
         UNIMPLEMENTED;
@@ -817,6 +848,12 @@ public:
         return makeLocalFposOffset(thisPartIdx-1, curStream->tell() + (const char *)_ptr - (const char *)curStream->peek(1, dummy));
     }
 
+    virtual unsigned __int64 makeFilePositionLocal(offset_t pos)
+    {
+        assertex(pos >= thisFileStartPos);
+        return makeLocalFposOffset(thisPartIdx-1, pos - thisFileStartPos);
+    }
+
     virtual const char * queryLogicalFilename(const void * row) 
     { 
         return f->queryLogicalFilename(thisPartIdx);

roxie/ccd/ccdkey.hpp

@@ -29,6 +29,7 @@ interface IDirectReader : public ISerialStream
     virtual IThorDiskCallback *queryThorDiskCallback() = 0;
     virtual ISimpleReadStream *querySimpleStream() = 0;
     virtual unsigned queryFilePart() const = 0;
+    virtual unsigned __int64 makeFilePositionLocal(offset_t pos) = 0;
 };
 
 interface IInMemoryIndexCursor : public IThorDiskCallback, public IIndexReadContext

roxie/ccd/ccdserver.cpp

@@ -20552,7 +20552,7 @@ class CRoxieServerXmlReadActivity : public CRoxieServerDiskReadBaseActivity, imp
     Owned<IXmlToRowTransformer> rowTransformer;
     Owned<IXMLParse> xmlParser;
     Owned<IColumnProvider> lastMatch;
-    unsigned __int64 localOffset;
+    unsigned __int64 fileoffset;
 public:
     IMPLEMENT_IINTERFACE;
     CRoxieServerXmlReadActivity(const IRoxieServerActivityFactory *_factory, IProbeManager *_probeManager, const RemoteActivityId &_remoteId, unsigned _numParts, bool _isLocal, bool _sorted, bool _maySkip, IInMemoryIndexManager *_manager)
@@ -20560,7 +20560,7 @@ public:
     {
         compoundHelper = NULL;
         readHelper = (IHThorXmlReadArg *)&helper;
-        localOffset = 0;
+        fileoffset = 0;
     }
 
     virtual void start(unsigned parentExtractSize, const byte *parentExtract, bool paused)
@@ -20590,22 +20590,22 @@ public:
 
     virtual void match(IColumnProvider &entry, offset_t startOffset, offset_t endOffset)
     {
-        localOffset = startOffset;
+        fileoffset = startOffset;
         lastMatch.set(&entry);
     }
 
     //interface IThorDiskCallback
     virtual unsigned __int64 getFilePosition(const void * row)
     {
-        return localOffset;
+        return fileoffset;
     }
     virtual unsigned __int64 getLocalFilePosition(const void * row)
     {
-        return localOffset;
+        return reader->makeFilePositionLocal(fileoffset);
     }
     virtual const char * queryLogicalFilename(const void * row)
     {
-        return varFileInfo ? varFileInfo->queryFileName() : NULL;
+        return reader->queryThorDiskCallback()->queryLogicalFilename(row);
     }
 
     virtual const void *nextRow()
@@ -20622,12 +20622,14 @@ public:
                 //call to next() will callback on the IXmlSelect interface
                 bool gotNext = false;
                 gotNext = xmlParser->next();
-                if (lastMatch)
+                if(!gotNext)
+                    eof = true;
+                else if (lastMatch)
                 {
                     RtlDynamicRowBuilder rowBuilder(rowAllocator);
                     unsigned sizeGot = rowTransformer->transform(rowBuilder, lastMatch, this);
                     lastMatch.clear();
-                    localOffset = 0;
+                    fileoffset = 0;
                     if (sizeGot)
                     {
                         OwnedConstRoxieRow ret = rowBuilder.finalizeRowClear(sizeGot); 
@@ -20647,8 +20649,6 @@ public:
                         return ret.getClear();
                     }
                 }
-                if(!gotNext) //currently slightly different behavior for json
-                    eof = true;
             }
             return NULL;
         }

system/jlib/jptree.cpp

@@ -6713,6 +6713,7 @@ class CPullJSONReader : public CJSONReaderBase<X>, implements IPullPTreeReader
     enum ParseStates { headerStart, nameStart, valueStart, itemStart, objAttributes, itemContent, itemEnd } state;
     bool endOfRoot;
     bool preReadItemName;
+    bool more;
     StringBuffer tag, value;
 
     void init()
@@ -6721,6 +6722,7 @@ class CPullJSONReader : public CJSONReaderBase<X>, implements IPullPTreeReader
         stateInfo = NULL;
         endOfRoot = false;
         preReadItemName = false;
+        more = true;
     }
 
     virtual void resetState()
@@ -6916,7 +6918,6 @@ public:
     }
     bool endNode(offset_t offset, bool notify=true)
     {
-        bool more = true;
         if (stack.ordinality()<2)
         {
             state = headerStart;
@@ -6932,7 +6933,7 @@ public:
         freeStateInfo.append(*stateInfo);
         stack.pop();
         stateInfo = (stack.ordinality()) ? &stack.tos() : NULL;
-        return more;
+        return true;
     }
 
     // IPullPTreeReader
@@ -6951,6 +6952,8 @@ public:
 
     virtual bool next()
     {
+        if (!more)
+            return false;
         checkStartReadNext();
         checkSkipWS();
         switch (state)

thorlcr/activities/fetch/thfetchslave.cpp

@@ -647,7 +647,7 @@ public:
         lastMatch = &lastMatches[filePartIndex];
         while (!lastMatch->get())
         {
-            if (!parser->next() && !lastMatch)
+            if (!parser->next())
             {
                 StringBuffer tmpStr;
                 throw MakeActivityException(this, 0, "%s", fetchStream->getPartName(filePartIndex, tmpStr).str());