Merge pull request #10741 from richardkchapman/roxie-fieldfilters

HPCC-18895 Roxie disk read to use new field filters

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

common/remote/sockfile.cpp

@@ -6303,9 +6303,9 @@ static StringBuffer basePath;
 static Owned<CSimpleInterface> serverThread;
 
 
-class RemoteFileTest : public CppUnit::TestFixture
+class RemoteFileSlowTest : public CppUnit::TestFixture
 {
-    CPPUNIT_TEST_SUITE(RemoteFileTest);
+    CPPUNIT_TEST_SUITE(RemoteFileSlowTest);
         CPPUNIT_TEST(testRemoteFilename);
         CPPUNIT_TEST(testStartServer);
         CPPUNIT_TEST(testBasicFunctionality);
@@ -6632,8 +6632,8 @@ protected:
     }
 };
 
-CPPUNIT_TEST_SUITE_REGISTRATION( RemoteFileTest );
-CPPUNIT_TEST_SUITE_NAMED_REGISTRATION( RemoteFileTest, "RemoteFileTests" );
+CPPUNIT_TEST_SUITE_REGISTRATION( RemoteFileSlowTest );
+CPPUNIT_TEST_SUITE_NAMED_REGISTRATION( RemoteFileSlowTest, "RemoteFileSlowTests" );
 
 
 #endif // _USE_CPPUNIT

common/thorhelper/thorxmlwrite.cpp

@@ -121,7 +121,7 @@ void CommonFieldProcessor::processEndRow(const RtlFieldInfo * field)
 
 
 //=============================================================================================
-
+// MORE - this function should probably move into IIndexReadContext interface rather than leaking ordinality() and item() out of that interface just for me
 void printKeyedValues(StringBuffer &out, IIndexReadContext *segs, IOutputMetaData *rowMeta)
 {
     unsigned totalKeyedSize = 0;

ecl/hqlcpp/hqlsource.cpp

@@ -3744,7 +3744,7 @@ const char * KeySelectorInfo::getFFOptions()
 //---------------------------------------------------------------------------------------------------------------------
 
 MonitorExtractor::MonitorExtractor(IHqlExpression * _tableExpr, HqlCppTranslator & _translator, int _numKeyableFields, bool _isDiskRead)
-    : translator(_translator), createValueSets(translator.queryOptions().createValueSets)
+    : translator(_translator), createValueSets(_isDiskRead || translator.queryOptions().createValueSets )
 { 
     tableExpr = _tableExpr;
 

roxie/ccd/ccdactivities.hpp

@@ -115,7 +115,6 @@ interface IInMemoryFileProcessor : extends IInterface
     virtual void abort() = 0;
 };
 
-IInMemoryFileProcessor *createKeyedGroupAggregateRecordProcessor(IInMemoryIndexCursor *cursor, RowAggregator &results, IHThorDiskGroupAggregateArg &helper);
-IInMemoryFileProcessor *createUnkeyedGroupAggregateRecordProcessor(IInMemoryIndexCursor *cursor, RowAggregator &results, IHThorDiskGroupAggregateArg &helper, IDirectReader *reader, ICodeContext *ctx, unsigned activityId);
+IInMemoryFileProcessor *createGroupAggregateRecordProcessor(RowAggregator &results, IHThorDiskGroupAggregateArg &helper, IDirectReader *reader);
 
 #endif

roxie/ccd/ccdcontext.cpp

@@ -2043,8 +2043,8 @@ protected:
             {
                 if (!workUnit)
                 	return factory->queryOnceContext(logctx);
-                // fall into...
             }
+            // fall into...
         case ResultSequenceInternal:
             {
                 CriticalBlock b(contextCrit);
@@ -2068,9 +2068,9 @@ protected:
         switch ((int) sequence)
         {
         case ResultSequenceOnce:
-        	if (!workUnit)
-        		return factory->queryOnceResultStore();
-        	// fall into...
+            if (!workUnit)
+                return factory->queryOnceResultStore();
+            // fall into...
         default:
             // No need to have separate stores for other temporaries...
             CriticalBlock b(contextCrit);

roxie/ccd/ccdfile.cpp

@@ -1689,21 +1689,31 @@ public:
 class CTranslatorSet : implements CInterfaceOf<ITranslatorSet>
 {
     IConstPointerArrayOf<IDynamicTransform> transformers;
+    IConstPointerArrayOf<IKeyTranslator> keyTranslators;
     IPointerArrayOf<IOutputMetaData> actualLayouts;
+    const RtlRecord &targetLayout;
     int targetFormatCrc = 0;
     bool anyTranslators = false;
 public:
-    CTranslatorSet(int _targetFormatCrc) : targetFormatCrc(_targetFormatCrc) {}
+    CTranslatorSet(const RtlRecord &_targetLayout, int _targetFormatCrc)
+    : targetLayout(_targetLayout), targetFormatCrc(_targetFormatCrc)
+    {}
 
-    void addTranslator(const IDynamicTransform *translator, IOutputMetaData *actualLayout)
+    void addTranslator(const IDynamicTransform *translator, const IKeyTranslator *keyTranslator, IOutputMetaData *actualLayout)
     {
         assertex(actualLayout);
         if (translator)
             anyTranslators = true;
         transformers.append(translator);
+        keyTranslators.append(keyTranslator);
         actualLayouts.append(actualLayout);
     }
 
+    virtual const RtlRecord &queryTargetFormat() const override
+    {
+        return targetLayout;
+    }
+
     virtual int queryTargetFormatCrc() const override
     {
         return targetFormatCrc;
@@ -1720,7 +1730,7 @@ public:
         return nullptr;
     }
 
-    virtual ISourceRowPrefetcher *getPrefetcher(unsigned subFile, bool addGroupFlag, ICodeContext *ctx, unsigned actId) const override
+    virtual ISourceRowPrefetcher *getPrefetcher(unsigned subFile, bool addGroupFlag) const override
     {
         IOutputMetaData *actualLayout = actualLayouts.item(subFile);
         assertex(actualLayout);
@@ -2093,8 +2103,9 @@ public:
     {
         // NOTE - projected and expected and anything fetched from them such as type info may reside in dynamically loaded (and unloaded)
         // query DLLs - this means it is not safe to include them in any sort of cache that might outlive the current query.
-        Owned<CTranslatorSet> result = new CTranslatorSet(formatCrc);
-        Owned<const IDynamicTransform> translator;
+        Owned<CTranslatorSet> result = new CTranslatorSet(expected->queryRecordAccessor(true), formatCrc);
+        Owned<const IDynamicTransform> translator;    // Translates rows from actual to projected
+        Owned<const IKeyTranslator> keyedTranslator;  // translate filter conditions from expected to actual
         int prevFormatCrc = 0;
         assertex(projected != nullptr);
         ForEachItemIn(idx, subFiles)
@@ -2108,12 +2119,14 @@ public:
                 {
                     actual = diskTypeInfo.item(idx);
                     translator.setown(createRecordTranslator(projected->queryRecordAccessor(true), actual->queryRecordAccessor(true)));
+                    keyedTranslator.setown(createKeyTranslator(actual->queryRecordAccessor(true), expected->queryRecordAccessor(true)));
                     if (!translator->canTranslate())
                         throw MakeStringException(ROXIE_MISMATCH, "Untranslatable record layout mismatch detected for file %s", subname);
                 }
                 else if (mode == IRecordLayoutTranslator::TranslateAlwaysECL)
                 {
                     translator.setown(createRecordTranslator(projected->queryRecordAccessor(true), expected->queryRecordAccessor(true)));
+                    keyedTranslator.setown(createKeyTranslator(actual->queryRecordAccessor(true), expected->queryRecordAccessor(true)));
                     if (!translator->canTranslate())
                         throw MakeStringException(ROXIE_MISMATCH, "Untranslatable record layout mismatch detected for file %s", subname);
                 }
@@ -2125,6 +2138,7 @@ public:
                     if (thisFormatCrc != prevFormatCrc)  // Check if same translation as last subfile
                     {
                         translator.clear();
+                        keyedTranslator.clear();
                         if (actual)
                         {
                             translator.setown(createRecordTranslator(projected->queryRecordAccessor(true), actual->queryRecordAccessor(true)));
@@ -2132,11 +2146,13 @@ public:
                         }
                         if (!translator || !translator->canTranslate())
                             throw MakeStringException(ROXIE_MISMATCH, "Untranslatable record layout mismatch detected for file %s", subname);
+                        if (translator->needsTranslate())
+                            keyedTranslator.setown(createKeyTranslator(actual->queryRecordAccessor(true), expected->queryRecordAccessor(true)));
                     }
                 }
                 prevFormatCrc = thisFormatCrc;
             }
-            result->addTranslator(LINK(translator), LINK(actual));
+            result->addTranslator(LINK(translator), LINK(keyedTranslator), LINK(actual));
         }
         return result.getClear();
     }
@@ -2330,7 +2346,7 @@ public:
         return ret.getClear();
     }
 
-    virtual IInMemoryIndexManager *getIndexManager(bool isOpt, unsigned channel, IOutputMetaData *preloadLayout, bool preload, int numKeys) const
+    virtual IInMemoryIndexManager *getIndexManager(bool isOpt, unsigned channel, IOutputMetaData *preloadLayout, bool preload) const
     {
         // MORE - I don't know that it makes sense to pass isOpt in to these calls
         // Failures to resolve will not be cached, only successes.
@@ -2340,9 +2356,9 @@ public:
         IInMemoryIndexManager *ret = indexMap.get(channel);
         if (!ret)
         {
-            ret = createInMemoryIndexManager(isOpt, lfn);
+            ret = createInMemoryIndexManager(preloadLayout->queryRecordAccessor(true), isOpt, lfn);
             Owned<IFileIOArray> files = getIFileIOArray(isOpt, channel);
-            ret->load(files, preloadLayout, preload, numKeys);   // note - files (passed in) are also channel specific
+            ret->load(files, preloadLayout, preload);   // note - files (passed in) are also channel specific
             indexMap.set(ret, channel);
         }
         return LINK(ret);

roxie/ccd/ccdfile.hpp

@@ -97,7 +97,7 @@ interface IResolvedFile : extends ISimpleSuperFileEnquiry
     virtual IKeyArray *getKeyArray(IDefRecordMeta *activityMeta, TranslatorArray *translators, bool isOpt, unsigned channel, IRecordLayoutTranslator::Mode allowFieldTranslation) const = 0;
     virtual IFilePartMap *getFileMap() const = 0;
     virtual unsigned getNumParts() const = 0;
-    virtual IInMemoryIndexManager *getIndexManager(bool isOpt, unsigned channel, IOutputMetaData *disklayout, bool preload, int numKeys) const = 0;
+    virtual IInMemoryIndexManager *getIndexManager(bool isOpt, unsigned channel, IOutputMetaData *disklayout, bool preload) const = 0;
     virtual offset_t getFileSize() const = 0;
 
     virtual const CDateTime &queryTimeStamp() const = 0;

roxie/ccd/ccdkey.hpp

@@ -21,46 +21,56 @@
 #include "eclhelper.hpp"
 #include "jfile.hpp"
 #include "rtlcommon.hpp"
+#include "rtlnewkey.hpp"
 
 interface IFileIOArray;
 interface ITranslatorSet;
 
 typedef IArrayOf<IKeySegmentMonitor> SegMonitorArray;
 
-interface IDirectReader : public ISerialStream
+/**
+ * IDirectStreamReader is used by CSV/XML readers. They bypass the record translation of the associated
+ * IDirectReader (this remains TBD at this point)
+ *
+ */
+interface IDirectStreamReader : extends ISerialStream, extends ISimpleReadStream
 {
-    virtual IThorDiskCallback *queryThorDiskCallback() = 0;
-    virtual ISimpleReadStream *querySimpleStream() = 0;
-    virtual unsigned queryFilePart() const = 0;
-    virtual unsigned __int64 makeFilePositionLocal(offset_t pos) = 0;
+    virtual unsigned queryFilePart() const = 0; // used by CSV
+    virtual unsigned __int64 makeFilePositionLocal(offset_t pos) = 0; // used by XML
+};
+
+/**
+ * IDirectReader is used by Roxie disk activities when the whole file needs to be scanned.
+ * There are in-memory and on-disk implementations, and ones that use indexes to seek directly to
+ * matching rows. Translated rows are returned.
+ *
+ */
+interface IDirectReader : extends IThorDiskCallback
+{
+    virtual IDirectStreamReader *queryDirectStreamReader() = 0;
     virtual const byte *nextRow() = 0;
-    virtual bool eog() const = 0;
     virtual void finishedRow() = 0;
-    virtual bool isTranslating() const = 0;
+    virtual void serializeCursorPos(MemoryBuffer &mb) const = 0;
+    virtual bool isKeyed() const = 0;
 };
 
-interface IInMemoryIndexCursor : public IThorDiskCallback, public IIndexReadContext
+class ScoredRowFilter : public RowFilter
 {
-    virtual void reset() = 0;
-    virtual bool selectKey() = 0;
-    virtual const void *nextMatch() = 0;
-    virtual bool isFiltered(const void *row) = 0;
-    virtual void serializeCursorPos(MemoryBuffer &mb) const = 0;
-    virtual void deserializeCursorPos(MemoryBuffer &mb) = 0;
+public:
+    unsigned scoreKey(const UnsignedArray &sortFields) const;
+    unsigned getMaxScore() const;
 };
 
 interface IInMemoryIndexManager : extends IInterface
 {
-    virtual void load(IFileIOArray *, IOutputMetaData *preloadLayout, bool preload, int numKeys) = 0;
+    virtual void load(IFileIOArray *, IOutputMetaData *preloadLayout, bool preload) = 0;
     virtual bool IsShared() const = 0;
-    virtual IInMemoryIndexCursor *createCursor(const RtlRecord &recInfo) = 0;
-    virtual IDirectReader *createReader(offset_t readPos, unsigned partNo, unsigned numParts, const ITranslatorSet *translators, ICodeContext *ctx, unsigned id) const = 0;
-    virtual void getTrackedInfo(const char *id, StringBuffer &xml) const = 0;
+    virtual IDirectReader *selectKey(ScoredRowFilter &filter, const ITranslatorSet *translators) const = 0;
+    virtual IDirectReader *selectKey(const char *sig, ScoredRowFilter &filter, const ITranslatorSet *translators) const = 0;
+    virtual IDirectReader *createReader(const RowFilter &postFilter, bool _grouped, offset_t readPos, unsigned partNo, unsigned numParts, const ITranslatorSet *translators) const = 0;
     virtual void setKeyInfo(IPropertyTree &indexInfo) = 0;
 };
 
-extern IInMemoryIndexManager *createInMemoryIndexManager(bool isOpt, const char *fileName);
-extern void reportInMemoryIndexStatistics(StringBuffer &reply, const char *filename, unsigned count);
-extern IInMemoryIndexManager *getEmptyIndexManager();
+extern IInMemoryIndexManager *createInMemoryIndexManager(const RtlRecord &recInfo, bool isOpt, const char *fileName);
 
 #endif

roxie/ccd/ccdprotocol.cpp

@@ -1350,7 +1350,7 @@ IHpccProtocolResponse *createProtocolResponse(const char *queryname, SafeSocket
 {
     StringAttr filter, tag;
     httpHelper.getResultFilterAndTag(filter, tag);
-    if (protocolFlags & HPCC_PROTOCOL_NATIVE_RAW || protocolFlags & HPCC_PROTOCOL_NATIVE_ASCII)
+    if ((protocolFlags & HPCC_PROTOCOL_NATIVE_RAW) || (protocolFlags & HPCC_PROTOCOL_NATIVE_ASCII))
         return new CHpccNativeProtocolResponse(queryname, client, MarkupFmt_Unknown, protocolFlags, false, logctx, xmlReadFlags, filter, tag);
     else if (httpHelper.queryResponseMlFormat()==MarkupFmt_JSON)
         return new CHpccJsonResponse(queryname, client, protocolFlags, httpHelper.isHttp(), logctx, xmlReadFlags, filter, tag);

roxie/ccd/ccdserver.cpp

@@ -7471,7 +7471,6 @@ class CRoxieServerDedupActivityFactory : public CRoxieServerActivityFactory
     bool compareAll;
     bool keepLeft;
     bool keepBest;
-    unsigned flags;
 
 public:
     CRoxieServerDedupActivityFactory(unsigned _id, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory, ThorActivityKind _kind, IPropertyTree &_graphNode)
@@ -21063,7 +21062,7 @@ public:
             assertex(outputLimit<=0x1000); // 32bit limit because MemoryBuffer/CMessageBuffers involved etc.
             outputLimitBytes = outputLimit * 0x100000;
         }
-        if (workunit != NULL || (results && protocol->getFlags() & HPCC_PROTOCOL_NATIVE_RAW))
+        if (workunit != NULL || (results && (protocol->getFlags() & HPCC_PROTOCOL_NATIVE_RAW)))
         {
             ensureRowAllocator();
             rowSerializer.setown(rowAllocator->createDiskSerializer(ctx->queryCodeContext()));
@@ -21084,7 +21083,7 @@ public:
                     result.append(row == NULL);
                 if (results)
                 {
-                    if (protocolFlags & HPCC_PROTOCOL_NATIVE_RAW && nativeResults)
+                    if ((protocolFlags & HPCC_PROTOCOL_NATIVE_RAW) && nativeResults)
                     {
                         char val = (char)(row == NULL);
                         nativeResults->appendRaw(sequence, 1, &val);
@@ -21447,10 +21446,9 @@ protected:
     unsigned __int64 stopAfter;
     Linked<IInMemoryIndexManager> manager;
     Linked<ITranslatorSet> translators;
-    Owned<IInMemoryIndexCursor> cursor;
+    ScoredRowFilter postFilter;
     Owned<IDirectReader> reader;
     bool eof;
-    bool isKeyed;
     bool variableFileName;
     bool isOpt;
     bool sorted;
@@ -21487,7 +21485,6 @@ public:
         compoundHelper = NULL;
         eof = false;
         rowLimit = (unsigned __int64) -1;
-        isKeyed = false;
         stopAfter = I64C(0x7FFFFFFFFFFFFFFF);
         diskSize.set(helper.queryDiskRecordSize()->querySerializedDiskMeta());
         isGrouped = diskSize.isGrouped();
@@ -21550,52 +21547,41 @@ public:
                     unsigned channel = isLocal ? factory->queryQueryFactory().queryChannel() : 0;
                     unsigned formatCrc = getFormatCrc(helper.getFormatCrc());
                     translators.setown(varFileInfo->getTranslators(formatCrc, helper.queryProjectedDiskRecordSize(), helper.queryDiskRecordSize(), getEnableFieldTranslation()));
-                    manager.setown(varFileInfo->getIndexManager(isOpt, channel, nullptr, false, 0));
+                    manager.setown(varFileInfo->getIndexManager(isOpt, channel, translators->queryActualLayout(0), false));
                 }
                 assertex(manager != NULL);
                 helper.createSegmentMonitors(this);
-                if (cursor)
-                {
-                    isKeyed = cursor->selectKey();
-                    cursor->reset();
-                }
-                if (!isKeyed)
-                {
-                    reader.setown(manager->createReader(0, 0, 1, translators, ctx->queryCodeContext(), activityId));
-                }
-                helper.setCallback(reader ? reader->queryThorDiskCallback() : cursor);
+                reader.setown(manager->selectKey(postFilter, translators));
+                if (!reader)
+                    reader.setown(manager->createReader(postFilter, isGrouped, 0, 0, 1, translators));
+                assertex(reader);
+                helper.setCallback(reader);
             }
         }
     }
 
     virtual void append(IKeySegmentMonitor *segment)
     {
-        if (!segment->isWild())
-        {
-            if (!cursor)
-                cursor.setown(manager->createCursor(diskSize.queryRecordAccessor(true)));
-            cursor->append(segment);
-        }
+        segment->Release();
+        throwUnexpected();
     }
 
     virtual void append(FFoption option, IFieldFilter * filter)
     {
-        if (!filter->isWild())
-        {
-            if (!cursor)
-                cursor.setown(manager->createCursor(diskSize.queryRecordAccessor(true)));
-            cursor->append(option, filter);
-        }
+        if (filter->isWild())
+            filter->Release();
+        else
+            postFilter.addFilter(*filter);
     }
 
     virtual unsigned ordinality() const
     {
-        return cursor ? cursor->ordinality() : 0;
+        throwUnexpected();
     }
 
     virtual IKeySegmentMonitor *item(unsigned idx) const
     {
-        return cursor ? cursor->item(idx) : 0;
+        throwUnexpected();
     }
 
     virtual void stop()
@@ -21615,8 +21601,6 @@ public:
         }
         varFileInfo.clear();
         eof = false;
-        if (cursor)
-            cursor->reset();
         reader.clear();
         CRoxieServerActivity::reset();
     }
@@ -21785,7 +21769,7 @@ public:
                 if (processed > rowLimit)
                 {
                     if (traceLevel > 4)
-                        DBGLOG("activityid = %d  isKeyed = %d  line = %d", activityId, isKeyed, __LINE__);
+                        DBGLOG("activityid = %d line = %d", activityId, __LINE__);
                     ReleaseRoxieRow(ret);
                     compoundHelper->onLimitExceeded();
                     throwUnexpected(); // onLimitExceeded is not supposed to return
@@ -21811,52 +21795,25 @@ public:
         }
         RtlDynamicRowBuilder rowBuilder(rowAllocator);
         unsigned transformedSize = 0;
-        if (isKeyed)
+        for (;;)
         {
-            for (;;)
+            const byte *nextRec = reader->nextRow();
+            if (nextRec)
             {
-                const void *nextCandidate = cursor->nextMatch();
-                if (!nextCandidate)
-                {
-                    eof = true;
-                    return NULL;
-                }
-                transformedSize = readHelper->transform(rowBuilder, nextCandidate);
+                transformedSize = readHelper->transform(rowBuilder, nextRec);
+                reader->finishedRow();
                 if (transformedSize)
                     break;
             }
-        }
-        else // use reader...
-        {
-            assertex(reader != NULL);
-            for (;;)
+            else if (isGrouped && lastGroupProcessed != processed)
             {
-                if (reader->eos())
-                {
-                    eof = true;
-                    return NULL;
-                }
-                const byte *nextRec = reader->nextRow();
-                if (cursor && cursor->isFiltered(nextRec))
-                    transformedSize = 0;
-                else
-                    transformedSize = readHelper->transform(rowBuilder, nextRec);
-                bool eog = isGrouped && reader->eog();
-                reader->finishedRow();
-                if (transformedSize)
-                {
-                    if (isGrouped)
-                        eogPending = eog;
-                    break;
-                }
-                else
-                {
-                    if (eog && (lastGroupProcessed != processed))
-                    {
-                        lastGroupProcessed = processed;
-                        return NULL;
-                    }
-                }
+                lastGroupProcessed = processed;
+                return NULL;
+            }
+            else
+            {
+                eof = true;
+                return nullptr;
             }
         }
         return rowBuilder.finalizeRowClear(transformedSize);
@@ -21870,6 +21827,7 @@ class CRoxieServerXmlReadActivity : public CRoxieServerDiskReadBaseActivity, imp
     Owned<IXMLParse> xmlParser;
     Owned<IColumnProvider> lastMatch;
     unsigned __int64 fileoffset;
+    IDirectStreamReader *streamReader = nullptr;
 public:
     IMPLEMENT_IINTERFACE_USING(CRoxieServerDiskReadBaseActivity)
     CRoxieServerXmlReadActivity(IRoxieSlaveContext *_ctx, const IRoxieServerActivityFactory *_factory, IProbeManager *_probeManager, const RemoteActivityId &_remoteId,
@@ -21890,11 +21848,13 @@ public:
         {
             rowTransformer.set(readHelper->queryTransformer());
             assertex(reader != NULL);
+            streamReader = reader->queryDirectStreamReader();
+            assertex(streamReader != NULL);
             OwnedRoxieString xmlIterator(readHelper->getXmlIteratorPath());
             if (factory->getKind()==TAKjsonread)
-                xmlParser.setown(createJSONParse(*reader->querySimpleStream(), xmlIterator, *this, (0 != (TDRxmlnoroot & readHelper->getFlags()))?ptr_noRoot:ptr_none, (readHelper->getFlags() & TDRusexmlcontents) != 0));
+                xmlParser.setown(createJSONParse(*streamReader, xmlIterator, *this, (0 != (TDRxmlnoroot & readHelper->getFlags()))?ptr_noRoot:ptr_none, (readHelper->getFlags() & TDRusexmlcontents) != 0));
             else
-                xmlParser.setown(createXMLParse(*reader->querySimpleStream(), xmlIterator, *this, (0 != (TDRxmlnoroot & readHelper->getFlags()))?ptr_noRoot:ptr_none, (readHelper->getFlags() & TDRusexmlcontents) != 0));
+                xmlParser.setown(createXMLParse(*streamReader, xmlIterator, *this, (0 != (TDRxmlnoroot & readHelper->getFlags()))?ptr_noRoot:ptr_none, (readHelper->getFlags() & TDRusexmlcontents) != 0));
         }
     }
 
@@ -21919,11 +21879,11 @@ public:
     }
     virtual unsigned __int64 getLocalFilePosition(const void * row)
     {
-        return reader->makeFilePositionLocal(fileoffset);
+        return streamReader->makeFilePositionLocal(fileoffset);
     }
     virtual const char * queryLogicalFilename(const void * row)
     {
-        return reader->queryThorDiskCallback()->queryLogicalFilename(row);
+        return reader->queryLogicalFilename(row);
     }
 
     virtual const void *nextRow()
@@ -21955,7 +21915,7 @@ public:
                         if (processed > rowLimit)
                         {
                             if (traceLevel > 4)
-                                DBGLOG("activityid = %d  isKeyed = %d  line = %d", activityId, isKeyed, __LINE__);
+                                DBGLOG("activityid = %d  line = %d", activityId, __LINE__);
                             readHelper->onLimitExceeded();
                             throwUnexpected(); // onLimitExceeded is not supposed to return
                         }
@@ -21986,6 +21946,7 @@ class CRoxieServerCsvReadActivity : public CRoxieServerDiskReadBaseActivity
     unsigned maxDiskSize;
     CSVSplitter csvSplitter;    
     unsigned __int64 localOffset;
+    IDirectStreamReader *streamReader = nullptr;
     const char *quotes;
     const char *separators;
     const char *terminators;
@@ -22015,11 +21976,14 @@ public:
         CRoxieServerDiskReadBaseActivity::start(parentExtractSize, parentExtract, paused);
         if (!useRemote())
         {
-            headerLines = csvInfo->queryHeaderLen(); 
-            if (headerLines && isLocal && reader->queryFilePart() != 1)
-                headerLines = 0;  // MORE - you could argue that if SINGLE not specified, should skip from all parts. But it would be painful since we have already concatenated and no-one else does...
             if (!eof)
             {
+                assertex(reader != nullptr);
+                streamReader = reader->queryDirectStreamReader();
+                assertex(streamReader != nullptr);
+                headerLines = csvInfo->queryHeaderLen();
+                if (headerLines && isLocal && streamReader->queryFilePart() != 1)
+                    headerLines = 0;  // MORE - you could argue that if SINGLE not specified, should skip from all parts. But it would be painful since we have already concatenated and no-one else does...
                 if (varFileInfo)
                 {
                     const IPropertyTree *options = varFileInfo->queryProperties();
@@ -22047,39 +22011,19 @@ public:
         {
             while (!eof)
             {
-                if (reader->eos())
-                {
-                    eof = true;
+                size32_t thisLineLength = csvSplitter.splitLine(streamReader, maxRowSize);
+                if (!thisLineLength)
                     break;
-                }
-                // MORE - there are rumours of a  csvSplitter that operates on a stream... if/when it exists, this should use it
-                size32_t rowSize = 4096; // MORE - make configurable
-                size32_t thisLineLength;
-                for (;;)
-                {
-                    size32_t avail;
-                    const void *peek = reader->peek(rowSize, avail);
-                    thisLineLength = csvSplitter.splitLine(avail, (const byte *)peek);
-                    if (thisLineLength < rowSize || avail < rowSize)
-                        break;
-                    if (rowSize == maxRowSize)
-                        throw MakeStringException(0, "File contained a line of length greater than %d bytes.", maxRowSize);
-                    if (rowSize >= maxRowSize/2)
-                        rowSize = maxRowSize;
-                    else
-                        rowSize += rowSize;
-                }
-
                 if (headerLines)
                 {
                     headerLines--;
-                    reader->skip(thisLineLength);
+                    streamReader->skip(thisLineLength);
                 }
                 else
                 {
                     RtlDynamicRowBuilder rowBuilder(rowAllocator);
                     unsigned transformedSize = readHelper->transform(rowBuilder, csvSplitter.queryLengths(), (const char * *)csvSplitter.queryData());
-                    reader->skip(thisLineLength);
+                    streamReader->skip(thisLineLength);
                     if (transformedSize)
                     {
                         OwnedConstRoxieRow ret = rowBuilder.finalizeRowClear(transformedSize);
@@ -22154,55 +22098,24 @@ public:
             return remote->nextRow();
         RtlDynamicRowBuilder rowBuilder(rowAllocator);
         unsigned transformedSize = 0;
-        if (isKeyed)
-        {
-            for (;;)
-            {
-                while (firstPending)
-                {
-                    const void *nextCandidate = cursor->nextMatch();
-                    if (!nextCandidate)
-                    {
-                        eof = true;
-                        return NULL;
-                    }
-                    if (normalizeHelper->first(nextCandidate))
-                    {
-                        firstPending = false;
-                        break;
-                    }
-                }
-                transformedSize = normalizeHelper->transform(rowBuilder);
-                firstPending = !normalizeHelper->next();
-                if (transformedSize)
-                    break;
-            }
-        }
-        else
+        for (;;)
         {
-            assertex(reader != NULL);
-            for (;;)
+            while (firstPending)
             {
-                while (firstPending)
+                const byte *nextRec = reader->nextRow();
+                if (!nextRec)
                 {
-                    if (reader->eos())
-                    {
-                        eof = true;
-                        return NULL;
-                    }
-                    const byte *nextRec = reader->nextRow();
-                    if (!cursor || !cursor->isFiltered(nextRec))
-                    {
-                        if (normalizeHelper->first(nextRec))
-                            firstPending = false;
-                    }
-                    reader->finishedRow();
+                    eof = true;
+                    return NULL;
                 }
-                transformedSize = normalizeHelper->transform(rowBuilder);
-                firstPending = !normalizeHelper->next();
-                if (transformedSize)
-                    break;
+                if (normalizeHelper->first(nextRec))
+                    firstPending = false;
+                reader->finishedRow();
             }
+            transformedSize = normalizeHelper->transform(rowBuilder);
+            firstPending = !normalizeHelper->next();
+            if (transformedSize)
+                break;
         }
         OwnedConstRoxieRow recBuffer = rowBuilder.finalizeRowClear(transformedSize);
         processed++;
@@ -22261,8 +22174,6 @@ class CRoxieServerDiskCountActivity : public CRoxieServerDiskAggregateBaseActivi
             return 1;
         else
         {
-            if (traceLevel > 4)
-                DBGLOG("activityid = %d  isKeyed = %d  line = %d", activityId, isKeyed, __LINE__);
             countHelper.onLimitExceeded();
             throwUnexpected(); // onLimitExceeded should always throw exception
         }
@@ -22319,47 +22230,22 @@ public:
             }
             else
             {
-                if (isKeyed)
+                for (;;)
                 {
-                    for (;;)
+                    const byte *nextRec = reader->nextRow();
+                    if (!nextRec)
+                        break;
+                    totalCount += countHelper.numValid(nextRec);
+                    reader->finishedRow();
+                    if (totalCount > rowLimit)
                     {
-                        const void *nextCandidate = cursor->nextMatch();
-                        if (!nextCandidate)
-                            break;
-                        totalCount += countHelper.numValid(nextCandidate);
-                        if (totalCount > rowLimit)
-                        {
-                            totalCount = getSkippedCount();
-                            break;
-                        }
-                        else if (totalCount >= choosenLimit)
-                        {
-                            totalCount = choosenLimit;
-                            break;
-                        }
+                        totalCount = getSkippedCount();
+                        break;
                     }
-                }
-                else
-                {
-                    assertex(reader != NULL);
-                    while (!reader->eos())
+                    else if (totalCount >= choosenLimit)
                     {
-                        const byte *nextRec = reader->nextRow();
-                        if (!cursor || !cursor->isFiltered(nextRec))
-                        {
-                            totalCount += countHelper.numValid(nextRec);
-                        }
-                        reader->finishedRow();
-                        if (totalCount > rowLimit)
-                        {
-                            totalCount = getSkippedCount();
-                            break;
-                        }
-                        else if (totalCount >= choosenLimit)
-                        {
-                            totalCount = choosenLimit;
-                            break;
-                        }
+                        totalCount = choosenLimit;
+                        break;
                     }
                 }
             }
@@ -22421,28 +22307,13 @@ public:
             aggregateHelper.clearAggregate(rowBuilder);
             if (helper.canMatchAny() && !eof)
             {
-                if (isKeyed)
-                {
-                    for (;;)
-                    {
-                        const void *next = cursor->nextMatch();
-                        if (!next)
-                            break;
-                        aggregateHelper.processRow(rowBuilder, next);
-                    }
-                }
-                else
+                for (;;)
                 {
-                    assertex(reader != NULL);
-                    while (!reader->eos())
-                    {
-                        const byte *nextRec = reader->nextRow();
-                        if (!cursor || !cursor->isFiltered(nextRec))
-                        {
-                            aggregateHelper.processRow(rowBuilder, nextRec);
-                        }
-                        reader->finishedRow();
-                    }
+                    const byte *nextRec = reader->nextRow();
+                    if (!nextRec)
+                        break;
+                    aggregateHelper.processRow(rowBuilder, nextRec);
+                    reader->finishedRow();
                 }
             }
             finalSize = meta.getRecordSize(rowBuilder.getSelf());
@@ -22506,10 +22377,7 @@ public:
         {
             if (helper.canMatchAny() && !eof)
             {
-                Owned<IInMemoryFileProcessor> processor = isKeyed ?
-                    createKeyedGroupAggregateRecordProcessor(cursor, resultAggregator, aggregateHelper) :
-                    createUnkeyedGroupAggregateRecordProcessor(cursor, resultAggregator, aggregateHelper, manager->createReader(0, 0, 1, translators, ctx->queryCodeContext(), activityId),
-                                                               ctx->queryCodeContext(), activityId);
+                Owned<IInMemoryFileProcessor> processor = createGroupAggregateRecordProcessor(resultAggregator, aggregateHelper, reader);
                 processor->doQuery(NULL, 0, 0, 0);
             }
         }
@@ -22576,7 +22444,7 @@ public:
                     unsigned channel = isLocal ? queryFactory.queryChannel() : 0;
                     unsigned formatCrc = getFormatCrc(helper->getFormatCrc());
                     translators.setown(datafile->getTranslators(formatCrc, helper->queryProjectedDiskRecordSize(), helper->queryDiskRecordSize(), getEnableFieldTranslation()));
-                    manager.setown(datafile->getIndexManager(isOpt, channel, translators->queryActualLayout(0), _graphNode.getPropBool("att[@name=\"preload\"]/@value", false), _graphNode.getPropInt("att[@name=\"_preloadSize\"]/@value", 0)));
+                    manager.setown(datafile->getIndexManager(isOpt, channel, translators->queryActualLayout(0), _graphNode.getPropBool("att[@name=\"preload\"]/@value", false)));
                     const IPropertyTree *options = datafile->queryProperties();
                     if (options)
                     {
@@ -22587,7 +22455,7 @@ public:
                     }
                 }
                 else
-                    manager.setown(getEmptyIndexManager());
+                    manager.setown(createInMemoryIndexManager(helper->queryProjectedDiskRecordSize()->queryRecordAccessor(true), true, nullptr));
             }
         }
         switch (kind)
@@ -25692,7 +25560,7 @@ public:
         if (keepLimit == 0) keepLimit = (unsigned)-1;
         getLimitType(joinFlags, limitFail, limitOnFail);
         cloneLeft = (joinFlags & JFtransformmatchesleft) != 0;
-        if (joinFlags & JFleftouter || limitOnFail)
+        if ((joinFlags & JFleftouter) || limitOnFail)
             createDefaultRight();
     }
 

roxie/ccd/ccdstate.cpp

@@ -2243,10 +2243,6 @@ private:
             {
                 reply.appendf("<clusterName id='%s'/>", roxieName.str());
             }
-            else if (stricmp(queryName, "control:getKeyInfo")==0)
-            {
-                reportInMemoryIndexStatistics(reply, control->queryProp("@id"), control->getPropInt("@count", 10));
-            }
             else if (stricmp(queryName, "control:getQueryXrefInfo")==0)
             {
                 getQueryInfo(control, reply, true, logctx);

roxie/ccd/ccdstate.hpp

@@ -104,9 +104,10 @@ interface IFileIOArray : extends IInterface
 interface ITranslatorSet : extends IInterface
 {
     virtual const IDynamicTransform *queryTranslator(unsigned subFile) const = 0;
-    virtual ISourceRowPrefetcher *getPrefetcher(unsigned subFile, bool addGroupedFlag, ICodeContext *ctx, unsigned actId) const = 0;
+    virtual ISourceRowPrefetcher *getPrefetcher(unsigned subFile, bool addGroupedFlag) const = 0;
     virtual IOutputMetaData *queryActualLayout(unsigned subFile) const = 0;
     virtual int queryTargetFormatCrc() const = 0;
+    virtual const RtlRecord &queryTargetFormat() const = 0;
     virtual bool isTranslating() const = 0;
 };
 

rtl/eclrtl/rtldynfield.cpp

@@ -25,6 +25,7 @@
 #include "rtldynfield.hpp"
 #include "rtlrecord.hpp"
 #include "rtlembed.hpp"
+#include "rtlnewkey.hpp"
 
 //#define TRACE_TRANSLATION
 #define VALIDATE_TYPEINFO_HASHES
@@ -1482,3 +1483,73 @@ extern ECLRTL_API IRowStream * transformRecord(IEngineRowAllocator * resultAlloc
     else
         return stream.getClear();
 }
+
+// A key translator allows us to transform a RowFilter that refers to src to one that refers to dest.
+// Basically just a map of those fields with matching types.
+
+class CKeyTranslator : public CInterfaceOf<IKeyTranslator>
+{
+public:
+    CKeyTranslator(const RtlRecord &destRecInfo, const RtlRecord &srcRecInfo)
+    {
+        for (unsigned idx = 0; idx < srcRecInfo.getNumFields(); idx++)
+        {
+            unsigned matchIdx = destRecInfo.getFieldNum(destRecInfo.queryName(idx));
+            if (matchIdx != -1)
+            {
+                const RtlTypeInfo *srcType = srcRecInfo.queryType(idx);
+                const RtlTypeInfo *destType = destRecInfo.queryType(idx);
+                if (!destType->equivalent(srcType))
+                    matchIdx = (unsigned) -2;
+            }
+            map.append(matchIdx);
+        }
+    }
+    virtual void describe() const override
+    {
+        ForEachItemIn(idx, map)
+        {
+            unsigned mapped = map.item(idx);
+            switch (mapped)
+            {
+            case (unsigned) -1: DBGLOG("No match for field %d", idx); break;
+            case (unsigned) -2: DBGLOG("Incompatible field match for field %d", idx); break;
+            default: DBGLOG("keyed field %d can map to field %d", idx, mapped); break;
+            }
+        }
+    }
+    virtual bool translate(RowFilter &filters) const override
+    {
+        bool mapNeeded = false;
+        unsigned numFields = filters.numFilterFields();
+        for (unsigned idx = 0; idx < numFields; idx++)
+        {
+            unsigned fieldNum = filters.queryFilter(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:
+                    filters.remapField(idx, mappedFieldNum);
+                    break;
+                }
+            }
+        }
+        if (mapNeeded)
+            filters.recalcFieldsRequired();
+        return mapNeeded;
+    }
+protected:
+    UnsignedArray map;
+};
+
+extern ECLRTL_API const IKeyTranslator *createKeyTranslator(const RtlRecord &_destRecInfo, const RtlRecord &_srcRecInfo)
+{
+    return new CKeyTranslator(_destRecInfo, _srcRecInfo);
+}
+
+

rtl/eclrtl/rtldynfield.hpp

@@ -109,7 +109,15 @@ interface IDynamicTransform : public IInterface
     virtual bool needsTranslate() const = 0;
 };
 
+class RowFilter;
+interface IKeyTranslator : public IInterface
+{
+    virtual void describe() const = 0;
+    virtual bool translate(RowFilter &filters) const = 0;
+};
+
 extern ECLRTL_API const IDynamicTransform *createRecordTranslator(const RtlRecord &_destRecInfo, const RtlRecord &_srcRecInfo);
+extern ECLRTL_API const IKeyTranslator *createKeyTranslator(const RtlRecord &_destRecInfo, const RtlRecord &_srcRecInfo);
 
 extern ECLRTL_API IRtlFieldTypeDeserializer *createRtlFieldTypeDeserializer(IThorIndexCallback *callback);
 

rtl/eclrtl/rtlfield.cpp

@@ -177,6 +177,62 @@ double RtlTypeInfoBase::getReal(const void * ptr) const
     return rtlStrToReal(len, value.getstr());
 }
 
+bool RtlTypeInfoBase::equivalent(const RtlTypeInfo *to) const
+{
+    if (to==this)
+        return true;
+    if (!to)
+        return false;
+    if (to->length != length || to->fieldType != fieldType)
+        return false;
+    if (queryLocale())
+    {
+        // do we permit a locale difference?
+    }
+    auto child = queryChildType();
+    if (child && !child->equivalent(to->queryChildType()))
+        return false;
+    auto fields = queryFields();
+    if (fields)
+    {
+        auto tofields = to->queryFields();
+        if (!tofields)
+            return false; // Should never happen
+        for (unsigned idx = 0; fields[idx]; idx++)
+        {
+            if (!fields[idx]->equivalent(tofields[idx]))
+                return false;
+        }
+    }
+    return true;
+}
+
+static bool strequivalent(const char *name1, const char *name2)
+{
+    if (name1)
+        return name2 && streq(name1, name2);
+    else
+        return name2==nullptr;
+}
+
+bool RtlFieldInfo::equivalent(const RtlFieldInfo *to) const
+{
+    if (to==this)
+        return true;
+    if (!to)
+        return false;
+    if (!strequivalent(name, to->name))
+        return false;
+    if (!strequivalent(xpath, to->xpath))
+        return false;
+    if (!type->equivalent(to->type))
+        return false;
+    // Initializer differences can be ignored
+    if (flags != to->flags)
+        return false;
+    return true;
+}
+
 const char * RtlTypeInfoBase::queryLocale() const 
 {
     return NULL; 

rtl/eclrtl/rtlfield.hpp

@@ -54,6 +54,7 @@ struct ECLRTL_API RtlTypeInfoBase : public RtlTypeInfo
     virtual bool canTruncate() const override { return false; }
     virtual bool canExtend(char &) const override { return false; }
     virtual bool canMemCmp() const override { return false; }
+    virtual bool equivalent(const RtlTypeInfo *) const override;
 
     virtual const char * queryLocale() const override;
     virtual const RtlFieldInfo * const * queryFields() const override;

rtl/eclrtl/rtlkey.hpp

@@ -295,6 +295,8 @@ public:
 
     virtual unsigned numRanges() const = 0;
     virtual int findForwardMatchRange(const RtlRow & row, unsigned & matchRange) const = 0;
+    virtual unsigned queryScore() const = 0;
+    virtual IFieldFilter *remap(unsigned newFieldIndex) const = 0;
 };
 
 //More types of IFieldFilter to come later

rtl/eclrtl/rtlnewkey.cpp

@@ -1046,6 +1046,8 @@ public:
     virtual int compareHighest(const RtlRow & left, unsigned range) const override;
     virtual int findForwardMatchRange(const RtlRow & row, unsigned & matchRange) const override;
 
+    virtual unsigned queryScore() const override;
+    virtual IFieldFilter *remap(unsigned newField) const override { return new SetFieldFilter(newField, values); }
 protected:
     Linked<IValueSet> values;
 };
@@ -1086,6 +1088,15 @@ int SetFieldFilter::findForwardMatchRange(const RtlRow & row, unsigned & matchRa
     return values->findForwardMatchRange(row.queryField(field), matchRange);
 }
 
+unsigned SetFieldFilter::queryScore() const
+{
+    // MORE - the score should probably depend on the number and nature of ranges too.
+    unsigned score = type.getMinSize();
+    if (!score)
+        score = 5;   // Arbitrary guess for average field length in a variable size field
+    return score;
+}
+
 IFieldFilter * createFieldFilter(unsigned fieldId, IValueSet * values)
 {
     return new SetFieldFilter(fieldId, values);
@@ -1155,6 +1166,11 @@ public:
         matchRange = 0;
         return true;
     }
+    virtual unsigned queryScore() const override
+    {
+        return 0;
+    }
+    virtual IFieldFilter *remap(unsigned newField) const override { return new WildFieldFilter(newField, type); }
 };
 
 IFieldFilter * createWildFieldFilter(unsigned fieldId, const RtlTypeInfo & type)
@@ -1172,14 +1188,18 @@ static int compareFieldFilters(IInterface * const * left, IInterface * const * r
     return leftFilter->queryFieldIndex() - rightFilter->queryFieldIndex();
 }
 
-void RowFilter::addFilter(IFieldFilter & filter)
+void RowFilter::addFilter(const IFieldFilter & filter)
 {
     //assertex(filter.queryField() == filters.ordinality()); //MORE - fill with wild filters and replace existing wild
     filters.append(filter);
+    unsigned fieldNum = filter.queryFieldIndex();
+    if (fieldNum >= numFieldsRequired)
+        numFieldsRequired = fieldNum+1;
 }
 
 bool RowFilter::matches(const RtlRow & row) const
 {
+    row.lazyCalcOffsets(numFieldsRequired);
     ForEachItemIn(i, filters)
     {
         if (!filters.item(i).matches(row))
@@ -1188,24 +1208,13 @@ bool RowFilter::matches(const RtlRow & row) const
     return true;
 }
 
-int RowFilter::compareRows(const RtlRow & left, const RtlRow & right) const
-{
-    ForEachItemIn(i, filters)
-    {
-        int rc = filters.item(i).compareRow(left, right);
-        if (rc != 0)
-            return rc;
-    }
-    return 0;
-}
-
-void RowFilter::extractKeyFilter(const RtlRecord & record, IArrayOf<IFieldFilter> & keyFilters) const
+void RowFilter::extractKeyFilter(const RtlRecord & record, IConstArrayOf<IFieldFilter> & keyFilters) const
 {
     if (!filters)
         return;
 
     // for an index must be in field order, and all values present
-    IArrayOf<IFieldFilter> temp;
+    IConstArrayOf<IFieldFilter> temp;
     ForEachItemIn(i, filters)
         temp.append(OLINK(filters.item(i)));
     temp.sort(compareFieldFilters);
@@ -1214,7 +1223,7 @@ void RowFilter::extractKeyFilter(const RtlRecord & record, IArrayOf<IFieldFilter
     unsigned curIdx=0;
     for (unsigned field = 0; field <= maxField; field++)
     {
-        IFieldFilter & cur = temp.item(curIdx);
+        const IFieldFilter & cur = temp.item(curIdx);
         if (field == cur.queryFieldIndex())
         {
             keyFilters.append(OLINK(cur));
@@ -1225,11 +1234,54 @@ void RowFilter::extractKeyFilter(const RtlRecord & record, IArrayOf<IFieldFilter
     }
 }
 
-//---------------------------------------------------------------------------------------------------------------------
+const IFieldFilter *RowFilter::findFilter(unsigned fieldNum) const
+{
+    ForEachItemIn(i, filters)
+    {
+        const IFieldFilter &field = filters.item(i);
+        if (field.queryFieldIndex() == fieldNum)
+            return &field;
+    }
+    return nullptr;
+}
 
+const IFieldFilter *RowFilter::extractFilter(unsigned fieldNum)
+{
+    ForEachItemIn(i, filters)
+    {
+        const IFieldFilter &field = filters.item(i);
+        if (field.queryFieldIndex() == fieldNum)
+        {
+            filters.remove(i, true);
+            return &field;
+        }
+    }
+    return nullptr;
+}
+
+void RowFilter::remove(unsigned idx)
+{
+    filters.remove(idx);
+}
+
+void RowFilter::recalcFieldsRequired()
+{
+    numFieldsRequired = 0;
+    ForEachItemIn(i, filters)
+    {
+        const IFieldFilter &field = filters.item(i);
+        if (field.queryFieldIndex() >= numFieldsRequired)
+            numFieldsRequired = field.queryFieldIndex()+1;
+    }
+}
 
+void RowFilter::remapField(unsigned filterIdx, unsigned newFieldNum)
+{
+    filters.replace(*filters.item(filterIdx).remap(newFieldNum), filterIdx);
+}
 
 
+//---------------------------------------------------------------------------------------------------------------------
 
 bool RowCursor::setRowForward(const byte * row)
 {
@@ -1329,53 +1381,6 @@ bool RowCursor::findNextRange(unsigned field)
 
 //---------------------------------------------------------------------------------------------
 
-class KeySearcher
-{
-public:
-    KeySearcher(const RtlRecord & _info, RowFilter & _filter, ISourceRowCursor * _rows) : cursor(_info, _filter), rows(_rows)
-    {
-    }
-
-    bool first()
-    {
-        rows->reset();
-        cursor.selectFirst();
-        return resolveValidRow();
-    }
-
-    bool next()
-    {
-        const byte * next = rows->next(); // MORE: Return a RtlRow?
-        if (!next)
-            return false;
-        if (cursor.setRowForward(next))
-            return true;
-        return resolveValidRow();
-    }
-
-    bool resolveValidRow()
-    {
-        for (;;)
-        {
-            if (cursor.noMoreMatches())
-                return false;
-            const byte * match;
-            match = rows->findNext(cursor); // more - return the row pointer to avoid recalculation
-            if (!match)
-                return false;
-            if (cursor.setRowForward(match))
-                return true;
-        }
-    }
-
-    const RtlRow & queryRow() const { return cursor.queryRow(); }
-
-protected:
-    ISourceRowCursor * rows = nullptr;
-    RowCursor cursor;
-};
-
-
 class InMemoryRows
 {
 public:
@@ -2312,12 +2317,10 @@ protected:
         KeySearcher searcher(source.queryRecord(), filter, &sourceCursor);
 
         StringBuffer matches;
-        if (searcher.first())
+        while (searcher.next())
         {
-            do
-            {
-                matches.append(searcher.queryRow().getInt(0)).append("|");
-            } while (searcher.next());
+            searcher.queryRow().lazyCalcOffsets(1);  // In unkeyed case we may not have calculated field 0 offset (though it is always going to be 0).
+            matches.append(searcher.queryRow().getInt(0)).append("|");
         }
 
         if (!streq(matches, expected))
@@ -2435,11 +2438,9 @@ protected:
             InMemoryRowCursor sourceCursor(source); // could be created by source.createCursor()
             KeySearcher searcher(source.queryRecord(), filter, &sourceCursor);
 
-            bool hasSearch = searcher.first();
-            while (hasSearch)
+            while (searcher.next())
             {
                 countKeyed++;
-                hasSearch = searcher.next();
             }
         }
         unsigned __int64 keyedMs = timeKeyed.elapsedNs();
@@ -2474,7 +2475,7 @@ protected:
         RowScanner scanner(source.queryRecord(), filter, rows);
 
         unsigned count = 0;
-        bool hasSearch = searcher.first();
+        bool hasSearch = searcher.next();
         bool hasScan = scanner.first();
         while (hasSearch && hasScan)
         {

rtl/eclrtl/rtlnewkey.hpp

@@ -28,19 +28,24 @@ BITMASK_ENUM(TransitionMask);
  * The RowFilter class represents a multiple-field filter of a row.
  */
 
-class RowFilter
+class ECLRTL_API RowFilter
 {
 public:
-    void addFilter(IFieldFilter & filter);
+    void addFilter(const IFieldFilter & filter);
     bool matches(const RtlRow & row) const;
 
-    void extractKeyFilter(const RtlRecord & record, IArrayOf<IFieldFilter> & keyFilters) const;
-    int compareRows(const RtlRow & left, const RtlRow & right) const;
+    void extractKeyFilter(const RtlRecord & record, IConstArrayOf<IFieldFilter> & keyFilters) const;
     unsigned numFilterFields() const { return filters.ordinality(); }
     const IFieldFilter & queryFilter(unsigned i) const { return filters.item(i); }
-
+    const IFieldFilter *findFilter(unsigned fieldIdx) const;
+    const IFieldFilter *extractFilter(unsigned fieldIdx);
+    unsigned getNumFieldsRequired() const { return numFieldsRequired; }
+    void remapField(unsigned filterIdx, unsigned newFieldNum);
+    void recalcFieldsRequired();
+    void remove(unsigned idx);
 protected:
-    IArrayOf<IFieldFilter> filters;
+    IConstArrayOf<IFieldFilter> filters;
+    unsigned numFieldsRequired = 0;
 };
 
 //This class represents the current set of values which have been matched in the filter sets.
@@ -49,7 +54,7 @@ protected:
 //Note: If any field is matching a range, then all subsequent fields must be matching the lowest possible filter range
 //      Therefore it is only necessary to keep track of the number of exactly matched fields, and which range the next
 //      field (if any) is matched against.
-class RowCursor
+class ECLRTL_API RowCursor
 {
 public:
     RowCursor(const RtlRecord & record, RowFilter & filter) : currentRow(record, nullptr)
@@ -142,7 +147,7 @@ protected:
     unsigned nextUnmatchedRange = 0;
     UnsignedArray matchedRanges;
     bool eos = false;
-    IArrayOf<IFieldFilter> filters; // for an index must be in field order, and all values present - more thought required
+    IConstArrayOf<IFieldFilter> filters; // for an index must be in field order, and all values present - more thought required
 };
 
 interface ISourceRowCursor
@@ -156,5 +161,60 @@ public:
     virtual void reset() = 0;
 };
 
+class ECLRTL_API KeySearcher : public CInterface
+{
+public:
+    KeySearcher(const RtlRecord & _info, RowFilter & _filter, ISourceRowCursor * _rows) : cursor(_info, _filter), rows(_rows)
+    {
+    }
+
+    void reset()
+    {
+        rows->reset();
+        firstPending = true;
+    }
+
+    bool next()
+    {
+        if (firstPending)
+        {
+            cursor.selectFirst();
+            firstPending = false;
+        }
+        else
+        {
+            const byte * next = rows->next(); // MORE: Return a RtlRow?
+            if (!next)
+                return false;
+            if (cursor.setRowForward(next))
+                return true;
+        }
+        return resolveValidRow();
+    }
+
+    bool resolveValidRow()
+    {
+        for (;;)
+        {
+            if (cursor.noMoreMatches())
+                return false;
+            const byte * match;
+            match = rows->findNext(cursor); // more - return the row pointer to avoid recalculation
+            if (!match)
+                return false;
+            if (cursor.setRowForward(match))
+                return true;
+        }
+    }
+
+    const RtlRow & queryRow() const { return cursor.queryRow(); }
+
+protected:
+    ISourceRowCursor * rows = nullptr;
+    RowCursor cursor;
+    bool firstPending = true;
+};
+
+
 
 #endif

rtl/eclrtl/rtlrecord.cpp

@@ -575,7 +575,7 @@ size32_t RtlRecord::calculateOffset(const void *_row, unsigned field) const
         unsigned numOffsets = getNumVarFields() + 1;
         size_t * variableOffsets = (size_t *)alloca(numOffsets * sizeof(size_t));
         RtlRow sourceRow(*this, nullptr, numOffsets, variableOffsets);
-        sourceRow.setRow(_row, field);
+        sourceRow.setRow(_row, field+1);
         return sourceRow.getOffset(field);
     }
     else
@@ -662,14 +662,16 @@ void RtlRow::getUtf8(size32_t & resultLen, char * & result, unsigned field) cons
     }
 }
 
-void RtlRow::setRow(const void * _row)
+void RtlRow::setRow(const void * _row, unsigned _numFieldsUsed)
 {
     row = (const byte *)_row;
     if (_row)
     {
-        info.calcRowOffsets(variableOffsets, _row);
+        numFieldsUsed = _numFieldsUsed;
+        if (numFieldsUsed)
+            info.calcRowOffsets(variableOffsets, _row, _numFieldsUsed);
 #if defined(_DEBUG) && defined(TRACE_ROWOFFSETS)
-        for (unsigned i = 0; i < info.getNumFields(); i++)
+        for (unsigned i = 0; i < info.getNumFields() && i < numFieldsUsed; i++)
         {
             printf("Field %d (%s) offset %d", i, info.queryName(i), (int) getOffset(i));
             if (getSize(i))
@@ -683,13 +685,22 @@ void RtlRow::setRow(const void * _row)
         }
 #endif
     }
+    else
+        numFieldsUsed = 0;
 }
 
-void RtlRow::setRow(const void * _row, unsigned _numFields)
+void RtlRow::lazyCalcOffsets(unsigned _numFieldsUsed) const
 {
-    row = (const byte *)_row;
-    if (_row)
-        info.calcRowOffsets(variableOffsets, _row, _numFields);
+    // This is a little iffy as it's not really const - but it clears up a lot of other code if you
+    // treat it as if it is. Logically it kind-of is, in that we are doing lazy-evaluation of the
+    // offsets but logically we are not creating information here.
+    // Another alternative would be to do the lazy eval in getOffset/getRecordSize ?
+    assert(row);
+    if (_numFieldsUsed > numFieldsUsed)
+    {
+        info.calcRowOffsets(variableOffsets, row, _numFieldsUsed); // MORE - could be optimized t oonly calc ones not previously calculated
+        numFieldsUsed = _numFieldsUsed;
+    }
 }
 
 RtlDynRow::RtlDynRow(const RtlRecord & _info, const void * optRow) : RtlRow(_info, optRow, _info.getNumVarFields()+1, new size_t[_info.getNumVarFields()+1])

rtl/eclrtl/rtlrecord.hpp

@@ -23,6 +23,7 @@
 #else
 #include <alloca.h>
 #endif
+#include <assert.h>
 
 #include "eclrtl_imp.hpp"
 #include "rtlds_imp.hpp"
@@ -262,21 +263,23 @@ public:
 
     size_t getOffset(unsigned field) const
     {
+        assert(field < numFieldsUsed);
         return info.getOffset(variableOffsets, field);
     }
 
     size_t getSize(unsigned field) const
     {
-        return info.getOffset(variableOffsets, field+1) - info.getOffset(variableOffsets, field);
+        return getOffset(field+1) - getOffset(field);
     }
 
     size_t getRecordSize() const
     {
+        assert(info.numFields <= numFieldsUsed);
         return info.getRecordSize(variableOffsets);
     }
 
-    void setRow(const void * _row);
-    void setRow(const void * _row, unsigned _numFields);
+    void setRow(const void * _row, unsigned _numFieldsUsed = (unsigned) -1);
+    void lazyCalcOffsets(unsigned _numFieldsUsed) const;
 
     const byte *queryRow() const
     {
@@ -290,6 +293,7 @@ public:
 protected:
     const RtlRecord & info;
     const byte * row;
+    mutable unsigned numFieldsUsed = 0;
     size_t * variableOffsets;       // [0 + 1 entry for each variable size field ]
 };
 

rtl/include/eclhelper.hpp

@@ -428,6 +428,7 @@ struct RtlTypeInfo : public RtlITypeInfo
     virtual bool canMemCmp() const = 0;
 
     virtual void doDelete() const = 0;  // Used in place of virtual destructor to allow constexpr constructors.
+    virtual bool equivalent(const RtlTypeInfo *other) const = 0;
 public:
     unsigned fieldType;
     unsigned length;                // for bitfield (int-size, # bits, bitoffset) << 16
@@ -475,6 +476,7 @@ struct RtlFieldInfo
     {
         return type->toXML(self, selfrow, this, target);
     }
+    bool equivalent(const RtlFieldInfo *to) const;
 };
 
 enum
@@ -2247,8 +2249,6 @@ interface ISteppingMeta
 //These were commoned up, but really they are completely different - so keep them separate
 interface IThorDiskCallback : extends IFilePositionProvider
 {
-    virtual unsigned __int64 getFilePosition(const void * row) = 0;
-    virtual unsigned __int64 getLocalFilePosition(const void * row) = 0;
     virtual const char * queryLogicalFilename(const void * row) = 0;
 };
 

system/jlib/jfile.cpp

@@ -5761,7 +5761,7 @@ public:
         eoinput = (_len==0);
     }
 
-    void reset(offset_t _offset, offset_t _len)
+    virtual void reset(offset_t _offset, offset_t _len) override
     {
         bufpos = 0;
         bufmax = 0;
@@ -5773,12 +5773,12 @@ public:
         eoinput = (_len==0);
     }
 
-    const void *peek(size32_t sz,size32_t &got)
+    virtual const void *peek(size32_t sz,size32_t &got) override
     {
         return dopeek(sz, got);
     }
 
-    void get(size32_t len, void * ptr)
+    virtual void get(size32_t len, void * ptr) override
     {
         size32_t cpy = bufmax-bufpos;
         if (cpy>len)
@@ -5792,7 +5792,7 @@ public:
         return getreadnext(len, (byte *)ptr+cpy);
     }
 
-    bool eos()
+    virtual bool eos() override
     {
         if (bufmax-bufpos)
             return false;
@@ -5800,7 +5800,7 @@ public:
         return dopeek(1,rd)==NULL;
     }
 
-    void skip(size32_t len)
+    virtual void skip(size32_t len) override
     {
         size32_t left = bufmax-bufpos;
         if (left>=len) {
@@ -5840,7 +5840,7 @@ public:
         throw MakeStringException(-1,"CFileSerialStream::skip read past end of stream");
     }
 
-    offset_t tell()
+    virtual offset_t tell() const override
     {
         return bufbase+bufpos;
     }
@@ -6004,7 +6004,7 @@ public:
         eoinput = false;
     }
 
-    void reset(offset_t _ofs, offset_t _len)
+    virtual void reset(offset_t _ofs, offset_t _len) override
     {
         offset_t fs = mmfile->fileSize();
         if ((_len!=(offset_t)-1)&&(fs>_len))
@@ -6012,6 +6012,7 @@ public:
         mmsize = (memsize_t)fs;
         mmofs = (memsize_t)((_ofs<fs)?_ofs:fs);
         mmsize = (memsize_t)fs;
+        eoinput = false;
     }
 
     CMemoryMappedSerialStream(const void *buf, memsize_t len, IFileSerialStreamCallback *_tally)
@@ -6023,7 +6024,7 @@ public:
         eoinput = false;
     }
 
-    const void *peek(size32_t sz,size32_t &got)
+    virtual const void *peek(size32_t sz,size32_t &got) override
     {
         memsize_t left = mmsize-mmofs;
         if (sz>left)
@@ -6035,7 +6036,7 @@ public:
         return mmbase+mmofs;
     }
 
-    void get(size32_t len, void * ptr)
+    virtual void get(size32_t len, void * ptr) override
     {
         memsize_t left = mmsize-mmofs;
         if (len>left) {
@@ -6049,12 +6050,12 @@ public:
         mmofs += len;
     }
 
-    bool eos()
+    virtual bool eos() override
     {
         return (mmsize<=mmofs);
     }
 
-    void skip(size32_t len)
+    virtual void skip(size32_t len) override
     {
         memsize_t left = mmsize-mmofs;
         if (len>left)
@@ -6064,16 +6065,11 @@ public:
         mmofs += len;
     }
 
-    offset_t tell()
+    virtual offset_t tell() const override
     {
         return mmofs;
     }
 
-    virtual void reset(offset_t _offset)
-    {
-        mmofs = (memsize_t)((_offset<mmsize)?_offset:mmsize);
-        eoinput = false;
-    }
 };
 
 ISerialStream *createFileSerialStream(IMemoryMappedFile *mmfile, offset_t ofs, offset_t flen, IFileSerialStreamCallback *callback)
@@ -6098,13 +6094,13 @@ public:
     {
     }
 
-    virtual const void *peek(size32_t sz,size32_t &got)
+    virtual const void *peek(size32_t sz,size32_t &got) override
     {
         got = buffer.remaining();
         return buffer.readDirect(0);
     }
 
-    virtual void get(size32_t len, void * ptr)
+    virtual void get(size32_t len, void * ptr) override
     {
         if (len>buffer.remaining()) {
             ERRLOG("CMemoryBufferSerialStream::get read past end of stream.4(%u,%u)",(unsigned)len,(unsigned)buffer.remaining());
@@ -6116,12 +6112,12 @@ public:
         memcpy(ptr,data,len);
     }
 
-    virtual bool eos()
+    virtual bool eos() override
     {
         return buffer.remaining() == 0;
     }
 
-    virtual void skip(size32_t len)
+    virtual void skip(size32_t len) override
     {
         if (len>buffer.remaining())
             throw MakeStringException(-1,"CMemoryBufferSerialStream::skip read past end of stream (%u,%u)",(unsigned)len,(unsigned)buffer.remaining());
@@ -6131,12 +6127,12 @@ public:
             tally->process(buffer.getPos()-len,len,data);
     }
 
-    virtual offset_t tell()
+    virtual offset_t tell() const override
     {
         return buffer.getPos();
     }
 
-    virtual void reset(offset_t _offset,offset_t _len)
+    virtual void reset(offset_t _offset,offset_t _len) override
     {
         size32_t ofs = (size32_t)_offset;
         assertex(ofs==_offset);

system/jlib/jfile.hpp

@@ -308,7 +308,7 @@ interface ISerialStream: extends IInterface
     virtual void get(size32_t len, void * ptr) = 0;                 // exception if no data available
     virtual bool eos() = 0;                                         // no more data
     virtual void skip(size32_t sz) = 0;
-    virtual offset_t tell() = 0;
+    virtual offset_t tell() const = 0;
     virtual void reset(offset_t _offset,offset_t _flen=(offset_t)-1) = 0;       // input stream has changed - restart reading
 };
 

system/jlib/jlib.hpp

@@ -139,6 +139,25 @@ public:
     inline bool zap(TYPE & obj, bool nodel=false) { assert(&obj); return IArray::zap(obj, nodel); }
 };
 
+template <class BTYPE>
+class IConstArrayOf : public IArray
+{
+    typedef const BTYPE TYPE;
+public:
+    inline TYPE & item(aindex_t pos) const        { return (TYPE &)IArray::item(pos); }
+    inline TYPE & popGet()                        { return (TYPE &)IArray::popGet(); }
+    inline TYPE & tos(void) const                 { return (TYPE &)IArray::tos(); }
+    inline TYPE & tos(aindex_t num) const         { return (TYPE &)IArray::tos(num); }
+    inline TYPE **getArray(aindex_t pos = 0)      { return (TYPE **)IArray::getArray(pos); }
+    inline TYPE **detach()                        { return (TYPE **)IArray::detach(); }
+    inline void append(TYPE& obj)                 { assert(&obj); IArray::append((BTYPE &) obj); }
+    inline void appendUniq(TYPE& obj)             { assert(&obj); IArray::appendUniq((BTYPE &) obj); }
+    inline void add(TYPE& obj, aindex_t pos)      { assert(&obj); IArray::add((BTYPE &) obj, pos); }
+    inline aindex_t find(TYPE & obj) const        { assert(&obj); return IArray::find((BTYPE &) obj); }
+    inline void replace(TYPE &obj, aindex_t pos, bool nodel=false) { assert(&obj); IArray::replace((BTYPE &) obj, pos, nodel); }
+    inline bool zap(TYPE & obj, bool nodel=false) { assert(&obj); return IArray::zap((BTYPE &) obj, nodel); }
+};
+
 template <class TYPE, class BASE>
 class IBasedArrayOf : public IArray
 {

testing/regress/ecl/sqfilt_keyed.ecl

@@ -18,9 +18,6 @@
 //Variable size comparisons only supported using the new field filters
 #option ('createValueSets', true);
 
-//more: need to implement field filters in roxie
-//noroxie
-
 //version multiPart=false
 
 import ^ as root;

testing/unittests/dalitests.cpp

@@ -788,9 +788,9 @@ CPPUNIT_TEST_SUITE_NAMED_REGISTRATION( CDaliTestsStress, "CDaliTestsStress" );
 
 // ================================================================================== UNIT TESTS
 
-class CDaliSDSTests : public CppUnit::TestFixture
+class CDaliSDSStressTests : public CppUnit::TestFixture
 {
-    CPPUNIT_TEST_SUITE(CDaliSDSTests);
+    CPPUNIT_TEST_SUITE(CDaliSDSStressTests);
         CPPUNIT_TEST(testInit);
         CPPUNIT_TEST(testSDSRW);
         CPPUNIT_TEST(testSDSSubs);
@@ -944,10 +944,10 @@ class CDaliSDSTests : public CppUnit::TestFixture
         return ret;
     }
 public:
-    CDaliSDSTests() : logctx(queryDummyContextLogger())
+    CDaliSDSStressTests() : logctx(queryDummyContextLogger())
     {
     }
-    ~CDaliSDSTests()
+    ~CDaliSDSStressTests()
     {
         daliClientEnd();
     }
@@ -1432,8 +1432,8 @@ public:
     }
 };
 
-CPPUNIT_TEST_SUITE_REGISTRATION( CDaliSDSTests );
-CPPUNIT_TEST_SUITE_NAMED_REGISTRATION( CDaliSDSTests, "CDaliSDSTests" );
+CPPUNIT_TEST_SUITE_REGISTRATION( CDaliSDSStressTests );
+CPPUNIT_TEST_SUITE_NAMED_REGISTRATION( CDaliSDSStressTests, "CDaliSDSStressTests" );
 
 // ================================================================================== UNIT TESTS
 
@@ -1517,9 +1517,9 @@ static void setupDFS(const IContextLogger &logctx, const char *scope, unsigned s
     }
 }
 
-class CDaliDFSTests : public CppUnit::TestFixture
+class CDaliDFSStressTests : public CppUnit::TestFixture
 {
-    CPPUNIT_TEST_SUITE(CDaliDFSTests);
+    CPPUNIT_TEST_SUITE(CDaliDFSStressTests);
         CPPUNIT_TEST(testInit);
         CPPUNIT_TEST(testGroups);
         CPPUNIT_TEST(testMultiCluster);
@@ -1553,10 +1553,10 @@ class CDaliDFSTests : public CppUnit::TestFixture
     const IContextLogger &logctx;
 
 public:
-    CDaliDFSTests() : logctx(queryDummyContextLogger())
+    CDaliDFSStressTests() : logctx(queryDummyContextLogger())
     {
     }
-    ~CDaliDFSTests()
+    ~CDaliDFSStressTests()
     {
         daliClientEnd();
     }
@@ -2250,8 +2250,8 @@ public:
     }
 };
 
-CPPUNIT_TEST_SUITE_REGISTRATION( CDaliDFSTests );
-CPPUNIT_TEST_SUITE_NAMED_REGISTRATION( CDaliDFSTests, "CDaliDFSTests" );
+CPPUNIT_TEST_SUITE_REGISTRATION( CDaliDFSStressTests );
+CPPUNIT_TEST_SUITE_NAMED_REGISTRATION( CDaliDFSStressTests, "CDaliDFSStressTests" );
 
 class CDaliDFSRetrySlowTests : public CppUnit::TestFixture
 {

testing/unittests/unittests.cpp

@@ -283,9 +283,9 @@ class InternalStatisticsTest : public CppUnit::TestFixture
 CPPUNIT_TEST_SUITE_REGISTRATION( InternalStatisticsTest );
 CPPUNIT_TEST_SUITE_NAMED_REGISTRATION( InternalStatisticsTest, "StatisticsTest" );
 
-class PtreeThreadingTest : public CppUnit::TestFixture
+class PtreeThreadingStressTest : public CppUnit::TestFixture
 {
-    CPPUNIT_TEST_SUITE( PtreeThreadingTest  );
+    CPPUNIT_TEST_SUITE( PtreeThreadingStressTest  );
         CPPUNIT_TEST(testContention);
     CPPUNIT_TEST_SUITE_END();
 
@@ -548,8 +548,8 @@ class PtreeThreadingTest : public CppUnit::TestFixture
     }
 };
 
-CPPUNIT_TEST_SUITE_REGISTRATION( PtreeThreadingTest );
-CPPUNIT_TEST_SUITE_NAMED_REGISTRATION( PtreeThreadingTest, "PtreeThreadingTest" );
+CPPUNIT_TEST_SUITE_REGISTRATION( PtreeThreadingStressTest );
+CPPUNIT_TEST_SUITE_NAMED_REGISTRATION( PtreeThreadingStressTest, "PtreeThreadingStressTest" );
 
 //MORE: This can't be included in jlib because of the dll dependency
 class StringBufferTest : public CppUnit::TestFixture