HPCC-21422 Fix problems with streamed datasets inside NOTHOR

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

ecl/hqlcpp/hqlcppds.cpp

@@ -2162,9 +2162,19 @@ void HqlCppTranslator::ensureDatasetFormat(BuildCtx & ctx, ITypeInfo * type, CHq
             OwnedITypeInfo streamedType = setStreamedAttr(type, true);
             OwnedHqlExpr call = bindFunctionCall(createRowStreamId, args, streamedType);
             buildTempExpr(ctx, call, tgt);
-            return;
         }
-        break;
+        return;
+    }
+
+    if (hasStreamedModifier(tgtType) && (format != FormatNatural))
+    {
+        OwnedHqlExpr serializedExpr = tgt.getTranslatedExpr();
+        buildTempExpr(ctx, serializedExpr, tgt, format);
+        tgtType = tgt.queryType();
+    }
+
+    switch (format)
+    {
     case FormatBlockedDataset:
         if (isArrayRowset(tgtType))
         {
@@ -2795,6 +2805,7 @@ void HqlCppTranslator::buildDatasetAssign(BuildCtx & ctx, const CHqlBoundTarget
         case no_translated:
         case no_null:
         case no_id2blob:
+            if (!hasStreamedModifier(exprType))
             {
                 IIdAtom * func = NULL;
                 if (!isArrayRowset(to))
@@ -2843,6 +2854,7 @@ void HqlCppTranslator::buildDatasetAssign(BuildCtx & ctx, const CHqlBoundTarget
                     return;
                 }
             }
+            break;
         }
     }
 

ecl/hqlcpp/hqlhtcpp.cpp

@@ -11658,7 +11658,7 @@ void HqlCppTranslator::doBuildStmtOutput(BuildCtx & ctx, IHqlExpression * expr)
     Owned<IWUResult> result = createDatasetResultSchema(seq, name, dataset->queryRecord(), xmlnsAttrs, true, false, 0);
 
     CHqlBoundExpr bound;
-    buildDataset(ctx, dataset, bound, FormatNatural);
+    buildDataset(ctx, dataset, bound, FormatBlockedDataset);
     OwnedHqlExpr count = getBoundCount(bound);
 
     HqlExprArray args;

testing/regress/ecl/key/stream.xml

@@ -0,0 +1,45 @@
+<Dataset name='Result 1'>
+ <Row><name>Mr Gavin</name></Row>
+ <Row><name>Mr John</name></Row>
+ <Row><name>Mr Bart</name></Row>
+</Dataset>
+<Dataset name='Result 2'>
+ <Row><name>Rev. Gavin</name></Row>
+ <Row><name>Rev. John</name></Row>
+ <Row><name>Rev. Bart</name></Row>
+</Dataset>
+<Dataset name='Result 3'>
+ <Row><name>Rev. Gavin</name></Row>
+ <Row><name>Rev. John</name></Row>
+ <Row><name>Rev. Bart</name></Row>
+</Dataset>
+<Dataset name='Result 4'>
+ <Row><name>Mr. Gavin</name></Row>
+ <Row><name>Mr. John</name></Row>
+ <Row><name>Mr. Bart</name></Row>
+</Dataset>
+<Dataset name='Result 5'>
+ <Row><name>Rev. Gavin</name></Row>
+ <Row><name>Rev. John</name></Row>
+ <Row><name>Rev. Bart</name></Row>
+</Dataset>
+<Dataset name='Result 6'>
+ <Row><name>Rev. Gavin</name></Row>
+ <Row><name>Rev. John</name></Row>
+ <Row><name>Rev. Bart</name></Row>
+</Dataset>
+<Dataset name='Result 7'>
+ <Row><name>Mr. Gavin</name></Row>
+ <Row><name>Mr. John</name></Row>
+ <Row><name>Mr. Bart</name></Row>
+</Dataset>
+<Dataset name='Result 8'>
+ <Row><name>Rev. Gavin</name></Row>
+ <Row><name>Rev. John</name></Row>
+ <Row><name>Rev. Bart</name></Row>
+</Dataset>
+<Dataset name='Result 9'>
+ <Row><name>Rev. Gavin</name></Row>
+ <Row><name>Rev. John</name></Row>
+ <Row><name>Rev. Bart</name></Row>
+</Dataset>

testing/regress/ecl/stream.ecl

@@ -0,0 +1,140 @@
+/*##############################################################################
+
+    HPCC SYSTEMS software Copyright (C) 2019 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.
+############################################################################## */
+
+//version inline=false
+//version inline=true
+
+import ^ as root;
+useInline := #IFDEFINED(root.inline, true);
+
+
+namesRecord := RECORD
+    STRING name;
+END;
+
+dataset(namesRecord) blockedNames(string prefix) := BEGINC++
+#define numElements(x) (sizeof(x)/sizeof(x[0]))
+    const char * const names[] = {"Gavin","John","Bart"};
+    unsigned len=0;
+    for (unsigned i = 0; i < numElements(names); i++)
+        len += sizeof(size32_t) + lenPrefix + strlen(names[i]);
+
+    byte * p = (byte *)rtlMalloc(len);
+    unsigned offset = 0;
+    for (unsigned j = 0; j < numElements(names); j++)
+    {
+        *(size32_t *)(p + offset) = lenPrefix + strlen(names[j]);
+        memcpy(p+offset+sizeof(size32_t), prefix, lenPrefix);
+        memcpy(p+offset+sizeof(size32_t)+lenPrefix, names[j], strlen(names[j]));
+        offset += sizeof(size32_t) + lenPrefix + strlen(names[j]);
+    }
+
+    __lenResult = len;
+    __result = p;
+ENDC++;
+
+
+_linkcounted_ dataset(namesRecord) linkedNames(string prefix) := BEGINC++
+
+#define numElements(x) (sizeof(x)/sizeof(x[0]))
+    const char * const names[] = {"Gavin","John","Bart"};
+    __countResult = numElements(names);
+    __result = _resultAllocator->createRowset(numElements(names));
+    for (unsigned i = 0; i < numElements(names); i++)
+    {
+        const char * name = names[i];
+        size32_t lenName = strlen(name);
+
+        RtlDynamicRowBuilder rowBuilder(_resultAllocator);
+        unsigned len = sizeof(size32_t) + lenPrefix + lenName;
+        byte * row = rowBuilder.ensureCapacity(len, NULL);
+        *(size32_t *)(row) = lenPrefix + lenName;
+        memcpy(row+sizeof(size32_t), prefix, lenPrefix);
+        memcpy(row+sizeof(size32_t)+lenPrefix, name, lenName);
+        __result[i] = (byte *)rowBuilder.finalizeRowClear(len);
+    }
+
+ENDC++;
+
+
+
+streamed dataset(namesRecord) streamedNames(string prefix) := BEGINC++
+
+#define numElements(x) (sizeof(x)/sizeof(x[0]))
+
+class StreamDataset : public RtlCInterface, implements IRowStream
+{
+public:
+    StreamDataset(IEngineRowAllocator * _resultAllocator, unsigned _lenPrefix, const char * _prefix)
+    : resultAllocator(_resultAllocator),lenPrefix(_lenPrefix), prefix(_prefix)
+    {
+        count = 0;
+    }
+    RTLIMPLEMENT_IINTERFACE
+
+    virtual const void *nextRow()
+    {
+        const char * const names[] = {"Gavin","John","Bart"};
+        if (count >= numElements(names))
+            return NULL;
+
+        const char * name = names[count++];
+        size32_t lenName = strlen(name);
+
+        RtlDynamicRowBuilder rowBuilder(resultAllocator);
+        unsigned len = sizeof(size32_t) + lenPrefix + lenName;
+        byte * row = rowBuilder.ensureCapacity(len, NULL);
+        *(size32_t *)(row) = lenPrefix + lenName;
+        memcpy(row+sizeof(size32_t), prefix, lenPrefix);
+        memcpy(row+sizeof(size32_t)+lenPrefix, name, lenName);
+        return rowBuilder.finalizeRowClear(len);
+    }
+    virtual void stop()
+    {
+        count = (unsigned)-1;
+    }
+
+
+protected:
+    Linked<IEngineRowAllocator> resultAllocator;
+    unsigned count;
+    unsigned lenPrefix;
+    const char * prefix;
+};
+
+#body
+    return new StreamDataset(_resultAllocator, lenPrefix, prefix);
+ENDC++;
+
+
+
+
+
+results := ordered(
+    output(blockedNames('Mr '));
+    output(blockedNames('Rev. '));
+    output(blockedNames('Rev. '));
+    output(linkedNames('Mr. '));
+    output(linkedNames('Rev. '));
+    output(linkedNames('Rev. '));
+    output(streamedNames('Mr. '));
+    output(streamedNames('Rev. '));
+    output(streamedNames('Rev. '));
+);
+
+action := IF(useInline, NOTHOR(results), results);
+action;