HPCC-9335 Smart stepping join continues to request rows after end of file

Pass information about whether the passed in seeks represent the final set of
desired matches to the slave node.

Stepping information not being returned from single graph result.

Added/tweaked tracing.

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

roxie/ccd/ccdactivities.cpp

@@ -3169,6 +3169,7 @@ protected:
     unsigned steppingLength;
     unsigned short numSkipFields;
     unsigned numSeeks;
+    bool seeksAreEof;
     bool lastRowCompleteMatch;
     CIndexTransformCallback callback;
 
@@ -3235,6 +3236,7 @@ public:
         keyprocessed = 0;
         numSkipFields = 0;
         lastRowCompleteMatch = true; // default is we only return complete matches....
+        seeksAreEof = false;
         steppingLength = 0;
         steppingRow = NULL;
         numSeeks = 0;
@@ -3247,6 +3249,8 @@ public:
             smartStepInfoValue += sizeof(unsigned short);
             unsigned flags = * (unsigned short *) smartStepInfoValue;
             smartStepInfoValue += sizeof(unsigned short);
+            seeksAreEof = * (bool *) smartStepInfoValue;
+            smartStepInfoValue += sizeof(bool);
             numSeeks = * (unsigned *) smartStepInfoValue;
             smartStepInfoValue += sizeof(unsigned);
             assertex(numSeeks); // Given that we put the first seek in here to there should always be at least one!
@@ -3257,12 +3261,15 @@ public:
                 logctx.CTXLOG("%d seek rows provided. mismatch(%d) readahead(%d) onlyfirst(%d)", numSeeks,
                              (int)stepExtra.returnMismatches(), (int)stepExtra.readAheadManyResults(), (int)stepExtra.onlyReturnFirstSeekMatch());
 
-                for (unsigned i = 0; i < numSeeks; i++)
+                if (logctx.queryTraceLevel() > 15)
                 {
-                    StringBuffer b;
-                    for (unsigned j = 0; j < steppingLength; j++)
-                        b.appendf("%02x ", steppingRow[i*steppingLength + j]);
-                    logctx.CTXLOG("Seek row %d: %s", i+1, b.str());
+                    for (unsigned i = 0; i < numSeeks; i++)
+                    {
+                        StringBuffer b;
+                        for (unsigned j = 0; j < steppingLength; j++)
+                            b.appendf("%02x ", steppingRow[i*steppingLength + j]);
+                        logctx.CTXLOG("Seek row %d: %s", i+1, b.str());
+                    }
                 }
             }
         }
@@ -3516,11 +3523,24 @@ public:
                         }
                         if (diff >= 0)
                         {
+                            if (diff > 0 && seeksAreEof)
+                            {
+                                assertex(!steppingRow);
+                                break;
+                            }
                             rowBuilder.ensureRow();
                             transformedSize = readHelper->transform(rowBuilder, keyRow);
                             callback.finishedRow();
                             if (transformedSize)
                             {
+                                if (logctx.queryTraceLevel() > 15)
+                                {
+                                    StringBuffer b;
+                                    for (unsigned j = 0; j < (steppingLength ? steppingLength : 6); j++)
+                                        b.appendf("%02x ", keyRow[steppingOffset + j]);
+                                    logctx.CTXLOG("Returning seek row %s", b.str());
+                                }
+
                                 // Did get a match
                                 processed++;
                                 if (limit && processed > limit)
@@ -3600,7 +3620,11 @@ public:
         if (tlk) // a very early abort can mean it is NULL.... MORE is this the right place to put it or should it be inside the loop??
         {
             if (logctx.queryTraceLevel() > 10 && !aborted)
+            {
                 logctx.CTXLOG("Indexread returning result set %d rows from %d seeks, %d scans, %d skips", processed-processedBefore, tlk->querySeeks(), tlk->queryScans(), tlk->querySkips());
+                if (steppingOffset)
+                    logctx.CTXLOG("Indexread return: steppingOffset %d, steppingRow %p, stepExtra.returnMismatches() %d",steppingOffset, steppingRow, (int) stepExtra.returnMismatches());
+            }
             logctx.noteStatistic(STATS_ACCEPTED, processed-processedBefore, 1);
             logctx.noteStatistic(STATS_REJECTED, skipped, 1);
         }

roxie/ccd/ccdserver.cpp

@@ -3825,7 +3825,7 @@ public:
 
     const void * nextSteppedGE(const void *seek, const void *rawSeek, unsigned numFields, unsigned seekLen, bool &wasCompleteMatch, const SmartStepExtra & stepExtra)
     {
-        if (activity.queryLogCtx().queryTraceLevel() > 10)
+        if (activity.queryLogCtx().queryTraceLevel() > 20)
         {
             StringBuffer recstr;
             unsigned i;
@@ -6017,6 +6017,18 @@ public:
         //helper already initialised from the gratherIterationUsage() call.
         iter.set(processor.connectIterationOutput(helper.querySequence(), probeManager, probes, this, 0));
     }
+
+    virtual IInputSteppingMeta * querySteppingMeta()
+    {
+        assertex(iter);
+        return iter->querySteppingMeta();
+    }
+
+    virtual const void * nextSteppedGE(const void * seek, unsigned numFields, bool &wasCompleteMatch, const SmartStepExtra & stepExtra)
+    {
+        assertex(iter);
+        return iter->nextSteppedGE(seek, numFields, wasCompleteMatch, stepExtra);
+    }
 };
 
 
@@ -21152,6 +21164,7 @@ public:
                 lookahead  = maxSeekLookahead;
             seeks->ensureFilled(seek, numFields, lookahead);
 
+            out.append(lookahead != seeks->ordinality()); // seeksAreEof flag
             unsigned serialized = 1; // rawseek is always serialized...
             unsigned patchLength = out.length();
             out.append(serialized);  // NOTE - we come back and patch with the actual value...
@@ -21187,6 +21200,7 @@ public:
         }
         else
         {
+            out.append(false);
             out.append(1);
             out.append(seekLen, rawSeek);
         }