HPCC-10456 Codegen support for streamed datasets from embedded languages

Work in progress, with hacks to get Richard moving.

Signed-off-by: Richard Chapman <rchapman@hpccsystems.com>
Signed-off-by: Gavin Halliday <gavin.halliday@lexisnexis.com>

ecl/hql/hqlatoms.cpp

@@ -34,6 +34,7 @@ IIdAtom * compileEmbeddedScriptId;
 IIdAtom * getEmbedContextId;
 IIdAtom * getBooleanResultId;
 IIdAtom * getDataResultId;
+IIdAtom * getDatasetResultId;
 IIdAtom * getRealResultId;
 IIdAtom * getSetResultId;
 IIdAtom * getSignedResultId;
@@ -439,6 +440,7 @@ MODULE_INIT(INIT_PRIORITY_HQLATOM)
     MAKEID(getEmbedContext);
     MAKEID(getBooleanResult);
     MAKEID(getDataResult);
+    MAKEID(getDatasetResult);
     MAKEID(getRealResult);
     MAKEID(getSetResult);
     MAKEID(getSignedResult);

ecl/hql/hqlatoms.hpp

@@ -36,6 +36,7 @@ extern HQL_API IIdAtom * failId;
 extern HQL_API IIdAtom * getEmbedContextId;
 extern HQL_API IIdAtom * getBooleanResultId;
 extern HQL_API IIdAtom * getDataResultId;
+extern HQL_API IIdAtom * getDatasetResultId;
 extern HQL_API IIdAtom * getRealResultId;
 extern HQL_API IIdAtom * getSetResultId;
 extern HQL_API IIdAtom * getSignedResultId;

ecl/hql/hqlgram2.cpp

@@ -894,7 +894,7 @@ IHqlExpression * HqlGram::processEmbedBody(const attribute & errpos, IHqlExpress
     result.setown(createLocationAnnotation(result.getClear(), errpos.pos));
 
     if (queryParametered())
-        return createWrapper(no_outofline, result.getClear());
+        return createWrapper(no_outofline, result.getClear(), createAttribute(contextAtom));    // MORE: this needs more thought
     return result.getClear();
 }
 

ecl/hqlcpp/hqlcpp.cpp

@@ -1034,7 +1034,10 @@ void CHqlBoundTarget::validate() const
             assertex(!length);
         }
         else if (isArrayRowset(queryType()))
-            assertex(count != NULL);
+        {
+            if (!hasStreamedModifier(queryType()))
+                assertex(count != NULL);
+        }
         else
         {
             assertex(length || queryType()->getTypeCode() == type_varstring || queryType()->getTypeCode() == type_varunicode);
@@ -11585,6 +11588,7 @@ void HqlCppTranslator::buildScriptFunctionDefinition(BuildCtx &funcctx, IHqlExpr
     funcctx.addQuoted("__ctx->callFunction();");
     IIdAtom * returnFunc;
     HqlExprArray retargs;
+    Owned<ITypeInfo> newReturnType;
     retargs.append(*LINK(ctxVar));
     switch (returnType->getTypeCode())
     {
@@ -11621,12 +11625,18 @@ void HqlCppTranslator::buildScriptFunctionDefinition(BuildCtx &funcctx, IHqlExpr
         retargs.append(*createIntConstant(returnType->queryChildType()->getSize()));
         break;
     }
+    case type_table:
+        {
+            returnFunc = getDatasetResultId;
+            newReturnType.set(returnType);
+            break;
+        }
     default:
         StringBuffer typeText;
         getFriendlyTypeStr(returnType, typeText);
         throwError1(HQLERR_EmbeddedTypeNotSupported_X, typeText.str());
     }
-    OwnedHqlExpr call = bindFunctionCall(returnFunc, retargs);
+    OwnedHqlExpr call = bindFunctionCall(returnFunc, retargs, newReturnType);
     doBuildUserFunctionReturn(funcctx, returnType, call);
 }
 

ecl/hqlcpp/hqlcppsys.ecl

@@ -855,6 +855,7 @@ const char * cppSystemText[]  = {
 
     "   boolean getBooleanResult() : method,entrypoint='getBooleanResult';",
     "   data getDataResult() : method,entrypoint='getDataResult';",
+    "   streamed dataset getDatasetResult() : method,entrypoint='getDatasetResult';",
     "   real getRealResult() : method,entrypoint='getRealResult';",
     "   integer getSignedResult() : method,entrypoint='getSignedResult';",
     "   string getStringResult() : method,entrypoint='getStringResult';",

ecl/hqlcpp/hqlhtcpp.cpp

@@ -3009,8 +3009,11 @@ void HqlCppTranslator::doBuildFunctionReturn(BuildCtx & ctx, ITypeInfo * type, I
     case type_table:
     case type_groupedtable:
     case type_row:
-        initBoundStringTarget(target, type, "__lenResult", "__result");
-        returnByReference = true;
+        if (!hasStreamedModifier(type))
+        {
+            initBoundStringTarget(target, type, "__lenResult", "__result");
+            returnByReference = true;
+        }
         break;
     case type_set:
         target.isAll.setown(createVariable("__isAllResult", makeBoolType()));

testing/ecl/streame.ecl

@@ -0,0 +1,41 @@
+/*##############################################################################
+
+    HPCC SYSTEMS software Copyright (C) 2012 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.
+############################################################################## */
+
+IMPORT Python;
+
+namesRecord := RECORD
+    STRING name;
+END;
+
+dataset(namesRecord) blockedNames(string prefix) := EMBED(Python)
+  return ["Gavin","John","Bart"]
+ENDEMBED;
+
+_linkcounted_ dataset(namesRecord) linkedNames(string prefix) := EMBED(Python)
+  return ["Gavin","John","Bart"]
+ENDEMBED;
+
+streamed dataset(namesRecord) streamedNames(string prefix) := EMBED(Python)
+  return ["Gavin","John","Bart"]
+ENDEMBED;
+
+titles := dataset(['', 'Mr. ', 'Rev. '], { string title });
+
+//output(normalize(titles, blockedNames(left.title), transform(right)));
+//output(normalize(titles, linkedNames(left.title), transform(right)));
+//output(normalize(titles, streamedNames(left.title), transform(right)));
+output(streamedNames('mr'));