Merge pull request #14258 from ghalliday/issue24898

HPCC-24898 Fix problem with orphaned results not being stopped

Reviewed-By: Mark Kelly <mark.kelly@lexisnexis.com>
Reviewed-By: Richard Chapman <rchapman@hpccsystems.com>

roxie/ccd/ccdserver.cpp

@@ -817,8 +817,10 @@ public:
 
     virtual bool isSink() const
     {
+        //If an internal result has as many dependencies (within the graph) as uses (which includes from outside the graph) then don't execute it unconditionally.
+        bool internalSpillAllUsesWithinGraph = (isInternal && dependentCount && dependentCount==usageCount);
         //only a sink if a root activity
-        return isRoot && !(isInternal && dependentCount && dependentCount==usageCount); // MORE - it's possible for this to get the answer wrong still, since usageCount does not include references from main procedure. Gavin?
+        return isRoot && !internalSpillAllUsesWithinGraph;
     }
 
     virtual void getEdgeProgressInfo(unsigned idx, IPropertyTree &edge) const
@@ -5315,6 +5317,17 @@ public:
     {
     }
 
+    //override execute() to ensure that start is nevr called on any input activities.
+    virtual void execute(unsigned parentExtractSize, const byte * parentExtract) override
+    {
+        CriticalBlock b(ecrit);
+        if (!executed)
+        {
+            executed = true;
+            stop();
+        }
+    }
+
     virtual void onExecute() override
     {
     }
@@ -21646,6 +21659,7 @@ public:
 class CRoxieServerWorkUnitWriteActivityFactory : public CRoxieServerInternalSinkFactory
 {
     bool isReread;
+    bool isUnused = false;
 
 public:
     CRoxieServerWorkUnitWriteActivityFactory(unsigned _id, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory, ThorActivityKind _kind, IPropertyTree &_graphNode, unsigned _usageCount, bool _isRoot)
@@ -21654,13 +21668,27 @@ public:
         isReread = usageCount > 0;
         Owned<IHThorWorkUnitWriteArg> helper = (IHThorWorkUnitWriteArg *) helperFactory();
         isInternal = (helper->getSequence()==ResultSequenceInternal);
+        isUnused = isInternal && (usageCount == 0);
     }
 
     virtual IRoxieServerActivity *createActivity(IRoxieSlaveContext *_ctx, IProbeManager *_probeManager) const
     {
+        if (isUnused && !CRoxieServerInternalSinkFactory::isSink())
+        {
+            if (_ctx->queryTraceLevel() > 2)
+                DBGLOG("Workunit write %u is unused - create null activity", id);
+            //Create a null sink activity that is always executed to ensure that stop() is called on the input.
+            return createRoxieServerNullSinkActivity(_ctx, this, _probeManager);
+        }
+
         return new CRoxieServerWorkUnitWriteActivity(_ctx, this, _probeManager, isReread, usageCount);
     }
 
+    virtual bool isSink() const
+    {
+        return isUnused || CRoxieServerInternalSinkFactory::isSink();
+    }
+
 };
 
 IRoxieServerActivityFactory *createRoxieServerWorkUnitWriteActivityFactory(unsigned _id, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory, ThorActivityKind _kind, IPropertyTree &_graphNode, unsigned _usageCount, bool _isRoot)

testing/regress/ecl/globalresult.ecl

@@ -0,0 +1,49 @@
+/*##############################################################################
+
+    HPCC SYSTEMS software Copyright (C) 2020 HPCC Systems®.
+
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+############################################################################## */
+
+
+//Ensure that references to temporaries outside of the graph are handled correctly
+ds := DATASET(10, transform({unsigned id}, SELF.id := COUNTER)) : GLOBAL(FEW);
+o1 := output(NOTHOR(count(ds)));
+
+
+idRec := {unsigned id};
+ds2base := DATASET(10, transform(idRec, SELF.id := COUNTER));
+
+//ds2 creates a global workunit write, which is then no longer needed once the child query is optimized.
+ds2 := ds2base : GLOBAL(FEW);
+
+
+ds3 := DATASET(20, transform({unsigned x, unsigned y}, SELF.x := COUNTER, SELF.y := 20));
+
+
+dsx(unsigned x) := DATASET(30, transform({DATASET(idRec) f1, unsigned f2}, SELF.f1 := ds2, SELF.f2 := x + COUNTER));
+
+ds3 t(ds3 l) := TRANSFORM
+    SELF.x := l.x;
+    SELF.y := COUNT(dsx(l.y));
+END;
+
+ds4 := project(nocombine(ds3), t(LEFT));
+
+sequential(
+    o1,
+    parallel(
+        output(ds4);
+        output(ds2base);
+    )
+);

testing/regress/ecl/key/globalresult.xml

@@ -0,0 +1,37 @@
+<Dataset name='Result 1'>
+ <Row><Result_1>10</Result_1></Row>
+</Dataset>
+<Dataset name='Result 2'>
+ <Row><x>1</x><y>30</y></Row>
+ <Row><x>2</x><y>30</y></Row>
+ <Row><x>3</x><y>30</y></Row>
+ <Row><x>4</x><y>30</y></Row>
+ <Row><x>5</x><y>30</y></Row>
+ <Row><x>6</x><y>30</y></Row>
+ <Row><x>7</x><y>30</y></Row>
+ <Row><x>8</x><y>30</y></Row>
+ <Row><x>9</x><y>30</y></Row>
+ <Row><x>10</x><y>30</y></Row>
+ <Row><x>11</x><y>30</y></Row>
+ <Row><x>12</x><y>30</y></Row>
+ <Row><x>13</x><y>30</y></Row>
+ <Row><x>14</x><y>30</y></Row>
+ <Row><x>15</x><y>30</y></Row>
+ <Row><x>16</x><y>30</y></Row>
+ <Row><x>17</x><y>30</y></Row>
+ <Row><x>18</x><y>30</y></Row>
+ <Row><x>19</x><y>30</y></Row>
+ <Row><x>20</x><y>30</y></Row>
+</Dataset>
+<Dataset name='Result 3'>
+ <Row><id>1</id></Row>
+ <Row><id>2</id></Row>
+ <Row><id>3</id></Row>
+ <Row><id>4</id></Row>
+ <Row><id>5</id></Row>
+ <Row><id>6</id></Row>
+ <Row><id>7</id></Row>
+ <Row><id>8</id></Row>
+ <Row><id>9</id></Row>
+ <Row><id>10</id></Row>
+</Dataset>