HPCC-9335 Track end of file from inputs used in stepped join

Signed-off-by: Gavin Halliday <gavin.halliday@lexisnexis.com>

common/thorhelper/thorstep.cpp

@@ -143,6 +143,7 @@ CSteppedInputLookahead::CSteppedInputLookahead(ISteppedInput * _input, IInputSte
     isPostFiltered = inputStepping ? inputStepping->hasPostFilter() : false;
     setRestriction(NULL, 0);
     lowestFrequencyInput = NULL;
+    eof = false;
 }
 
 CSteppedInputLookahead::~CSteppedInputLookahead()
@@ -158,7 +159,12 @@ const void * CSteppedInputLookahead::nextInputRow()
 {
     if (readAheadRows.ordinality())
         return readAheadRows.dequeue();
-    return input->nextInputRow();
+    if (eof)
+        return NULL;
+    const void * ret = input->nextInputRow();
+    if (!ret)
+        eof = true;
+    return ret;
 }
     
 const void * CSteppedInputLookahead::nextInputRowGE(const void * seek, unsigned numFields, bool & wasCompleteMatch, const SmartStepExtra & stepExtra)
@@ -172,7 +178,12 @@ const void * CSteppedInputLookahead::nextInputRowGE(const void * seek, unsigned
             return (void *)next.getClear();
         }
     }
-    return input->nextInputRowGE(seek, numFields, wasCompleteMatch, stepExtra);
+    if (eof)
+        return NULL;
+    const void * ret = input->nextInputRowGE(seek, numFields, wasCompleteMatch, stepExtra);
+    if (!ret)
+        eof = true;
+    return ret;
 }
 
 void CSteppedInputLookahead::ensureFilled(const void * seek, unsigned numFields, unsigned maxcount)
@@ -204,6 +215,9 @@ void CSteppedInputLookahead::ensureFilled(const void * seek, unsigned numFields,
         }
     }
 
+    if (eof)
+        return;
+
     //Return mismatches is selected because we don't want it to seek exact matches beyond the last seek position
     unsigned flags = (SSEFreturnMismatches & ~stepFlagsMask) | stepFlagsValue;
     SmartStepExtra inputStepExtra(flags, lowestFrequencyInput);
@@ -213,7 +227,10 @@ void CSteppedInputLookahead::ensureFilled(const void * seek, unsigned numFields,
         bool wasCompleteMatch = true;
         const void * next = input->nextInputRowGE(seek, numFields, wasCompleteMatch, inputStepExtra);
         if (!next)
+        {
+            eof = true;
             break;
+        }
         //wasCompleteMatch can be false if we've just read the last row returned from a block of reads, 
         //but if so the next read request will do another blocked read, so just ignore this one.
         if (wasCompleteMatch)

common/thorhelper/thorstep.ipp

@@ -255,6 +255,7 @@ protected:
     bool paranoid;
     bool readAheadRowIsExactMatch;
     bool isPostFiltered;
+    bool eof;
 };
 
 class DummySteppedInput : public CSteppedInputLookahead