HPCC-14050 Fix issue with child filter + varying canMatchAny()

A filter with a canMatchAny() evaluating to false, caused the
graph to be truncated at that point. If these appear in a child
query and the canMatchAny() condition varies from run to run, it
causes the graph to incorrectly connected, resulting in an
assert being hit.

Signed-off-by: Jake Smith <jake.smith@lexisnexis.com>

thorlcr/graph/thgraph.cpp

@@ -373,7 +373,7 @@ CGraphElementBase::CGraphElementBase(CGraphBase &_owner, IPropertyTree &_xgmml)
     whichBranch = (unsigned)-1;
     whichBranchBitSet.setown(createThreadSafeBitSet());
     newWhichBranch = false;
-    isEof = false;
+    hasNullInput = false;
     log = true;
     sentActInitData.setown(createThreadSafeBitSet());
 }
@@ -476,7 +476,7 @@ void CGraphElementBase::doconnect()
     {
         CIOConnection *io = connectedInputs.item(i);
         if (io)
-            io->connect(i, activity);
+            io->connect(i, queryActivity());
     }
 }
 
@@ -619,7 +619,7 @@ bool CGraphElementBase::prepareContext(size32_t parentExtractSz, const byte *par
                 return false;
         }
         whichBranch = (unsigned)-1;
-        isEof = false;
+        hasNullInput = false;
         alreadyUpdated = false;
         switch (getKind())
         {
@@ -688,16 +688,16 @@ bool CGraphElementBase::prepareContext(size32_t parentExtractSz, const byte *par
                 switch (getKind())
                 {
                     case TAKfilter:
-                        isEof = !((IHThorFilterArg *)baseHelper.get())->canMatchAny();
+                        hasNullInput = !((IHThorFilterArg *)baseHelper.get())->canMatchAny();
                         break;
                     case TAKfiltergroup:
-                        isEof = !((IHThorFilterGroupArg *)baseHelper.get())->canMatchAny();
+                        hasNullInput = !((IHThorFilterGroupArg *)baseHelper.get())->canMatchAny();
                         break;
                     case TAKfilterproject:
-                        isEof = !((IHThorFilterProjectArg *)baseHelper.get())->canMatchAny();
+                        hasNullInput = !((IHThorFilterProjectArg *)baseHelper.get())->canMatchAny();
                         break;
                 }
-                if (isEof)
+                if (hasNullInput)
                     return true;
                 break;
             }
@@ -831,45 +831,45 @@ void CGraphElementBase::createActivity(size32_t parentExtractSz, const byte *par
                 break;
             }
             default:
-                if (!isEof)
+                if (!hasNullInput)
                 {
                     ForEachItemIn(i, inputs)
                     {
                         CGraphElementBase *input = inputs.item(i)->activity;
                         input->createActivity(parentExtractSz, parentExtract);
                     }
-                }
-                onCreate();
-                if (isDiskInput(getKind()))
-                    onStart(parentExtractSz, parentExtract);
-                ForEachItemIn(i2, inputs)
-                {
-                    CIOConnection *inputIO = inputs.item(i2);
-                    loop
+                    onCreate();
+                    if (isDiskInput(getKind()))
+                        onStart(parentExtractSz, parentExtract);
+                    ForEachItemIn(i2, inputs)
                     {
-                        CGraphElementBase *input = inputIO->activity;
-                        switch (input->getKind())
+                        CIOConnection *inputIO = inputs.item(i2);
+                        loop
                         {
-                            case TAKif:
-                            case TAKcase:
+                            CGraphElementBase *input = inputIO->activity;
+                            switch (input->getKind())
                             {
-                                if (input->whichBranch >= input->getInputs()) // if, will have TAKnull activity, made at create time.
+                                case TAKif:
+                                case TAKcase:
                                 {
-                                    input = NULL;
+                                    if (input->whichBranch >= input->getInputs()) // if, will have TAKnull activity, made at create time.
+                                    {
+                                        input = NULL;
+                                        break;
+                                    }
+                                    inputIO = input->inputs.item(input->whichBranch);
+                                    assertex(inputIO);
                                     break;
                                 }
-                                inputIO = input->inputs.item(input->whichBranch);
-                                assertex(inputIO);
-                                break;
+                                default:
+                                    input = NULL;
+                                    break;
                             }
-                            default:
-                                input = NULL;
+                            if (!input)
                                 break;
                         }
-                        if (!input)
-                            break;
+                        connectInput(i2, inputIO->activity, inputIO->index);
                     }
-                    connectInput(i2, inputIO->activity, inputIO->index);
                 }
                 initActivity();
                 break;
@@ -1590,14 +1590,14 @@ public:
     CGraphTraverseConnectedIterator(CGraphBase &graph) : CGraphTraverseIteratorBase(graph) { }
     virtual bool next()
     {
-        if (cur->isEof)
+        if (cur->hasNullInput)
         {
             do
             {
                 if (!popNext())
                     return false;
             }
-            while (cur->isEof);
+            while (cur->hasNullInput);
         }
         else
             setNext(cur->connectedInputs);

thorlcr/graph/thgraph.hpp

@@ -260,7 +260,7 @@ public:
 
     const void *queryFindParam() const { return &queryId(); } // for SimpleHashTableOf
 
-    bool alreadyUpdated, isEof, newWhichBranch;
+    bool alreadyUpdated, hasNullInput, newWhichBranch;
     EclHelperFactory helperFactory;
 
     CIOConnectionArray inputs, outputs, connectedInputs, connectedOutputs;
@@ -330,7 +330,6 @@ public:
     IHThorArg *queryHelper() const { return baseHelper; }
 
     IPropertyTree &queryXGMML() const { return *xgmml; }
-    CActivityBase *queryActivity() const { return activity; }
     const activity_id &queryOwnerId() const { return ownerId; }
     void createActivity(size32_t parentExtractSz, const byte *parentExtract);
 //
@@ -343,6 +342,7 @@ public:
     }
     virtual bool prepareContext(size32_t parentExtractSz, const byte *parentExtract, bool checkDependencies, bool shortCircuit, bool async);
 //
+    virtual CActivityBase *queryActivity() { return activity; }
     virtual void initActivity();
     virtual CActivityBase *factory(ThorActivityKind kind) { assertex(false); return NULL; }
     virtual CActivityBase *factory() { return factory(getKind()); }

thorlcr/graph/thgraphmaster.cpp

@@ -585,10 +585,10 @@ bool CMasterGraphElement::checkUpdate()
 void CMasterGraphElement::initActivity()
 {
     CriticalBlock b(crit);
-    bool first = (NULL == activity);
+    bool first = (NULL == queryActivity());
     CGraphElementBase::initActivity();
-    if (first || activity->needReInit())
-        ((CMasterActivity *)activity.get())->init();
+    if (first || queryActivity()->needReInit())
+        ((CMasterActivity *)queryActivity())->init();
 }
 
 void CMasterGraphElement::doCreateActivity(size32_t parentExtractSz, const byte *parentExtract)
@@ -627,7 +627,7 @@ void CMasterGraphElement::doCreateActivity(size32_t parentExtractSz, const byte
 
 void CMasterGraphElement::slaveDone(size32_t slaveIdx, MemoryBuffer &mb)
 {
-    ((CMasterActivity *)activity.get())->slaveDone(slaveIdx, mb);
+    ((CMasterActivity *)queryActivity())->slaveDone(slaveIdx, mb);
 }
 
 

thorlcr/slave/slave.cpp

@@ -263,6 +263,8 @@ class CGenericSlaveGraphElement : public CSlaveGraphElement
 {
     bool wuidread2diskread; // master decides after interrogating result and sneaks in info before slave creates
     StringAttr wuidreadFilename;
+    Owned<CActivityBase> nullActivity;
+    CriticalSection nullActivityCs;
 public:
     CGenericSlaveGraphElement(CGraphBase &_owner, IPropertyTree &xgmml) : CSlaveGraphElement(_owner, xgmml)
     {
@@ -287,6 +289,18 @@ public:
         }
         haveCreateCtx = true;
     }
+    virtual CActivityBase *queryActivity()
+    {
+        if (hasNullInput)
+        {
+            CriticalBlock b(nullActivityCs);
+            if (!nullActivity)
+                nullActivity.setown(createNullSlave(this));
+            return nullActivity;
+        }
+        else
+            return activity;
+    }
     virtual CActivityBase *factory(ThorActivityKind kind)
     {
         CActivityBase *ret = NULL;