HPCC-10456 Codegen support for streamed datasets from embedded languages

Support datarow parameters and other fixes

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

ecl/hql/hqlatoms.cpp

@@ -294,6 +294,7 @@ IAtom * parameterAtom;
 IAtom * partitionAtom;
 IAtom * partitionLeftAtom;
 IAtom * partitionRightAtom;
+IAtom * passParameterMetaAtom;
 IAtom * _payload_Atom;
 IAtom * persistAtom;
 IAtom * physicalFilenameAtom;
@@ -704,6 +705,7 @@ MODULE_INIT(INIT_PRIORITY_HQLATOM)
     MAKEATOM(partition);
     partitionLeftAtom = createLowerCaseAtom("partition left");
     partitionRightAtom = createLowerCaseAtom("partition right");
+    MAKEATOM(passParameterMeta);
     MAKESYSATOM(payload);
     MAKEATOM(persist);
     MAKEATOM(physicalFilename);

ecl/hql/hqlatoms.hpp

@@ -297,6 +297,7 @@ extern HQL_API IAtom * parameterAtom;
 extern HQL_API IAtom * partitionAtom;
 extern HQL_API IAtom * partitionLeftAtom;
 extern HQL_API IAtom * partitionRightAtom;
+extern HQL_API IAtom * passParameterMetaAtom;
 extern HQL_API IAtom * payloadAtom;
 extern HQL_API IAtom * _payload_Atom;
 extern HQL_API IAtom * persistAtom;

ecl/hql/hqlgram.y

@@ -803,6 +803,29 @@ explicitDictionaryType
     | userTypedefDictionary
     ;
 
+explicitRowType
+    : explicitRowType1
+    | LINKCOUNTED explicitRowType1
+                        {
+                            $$.setType(setLinkCountedAttr($2.getType(), true));
+                            $$.setPosition($1);
+                        }
+    ;
+
+explicitRowType1
+    : ROW               {
+                            IHqlExpression* record = queryNullRecord();
+                            $$.setType(makeRowType(record->getType()));
+                            $$.setPosition($1);
+                        }
+    | ROW '(' recordDef ')'
+                        {
+                            OwnedHqlExpr record = $3.getExpr();
+                            $$.setType(makeRowType(record->getType()));
+                            $$.setPosition($1);
+                        }
+    ;
+
 
 transformType
     : TRANSFORM '(' recordDef ')'
@@ -865,17 +888,7 @@ paramType
                         }
     | explicitDatasetType
     | explicitDictionaryType
-    | ROW               {
-                            IHqlExpression* record = queryNullRecord();
-                            $$.setType(makeRowType(record->getType()));
-                            $$.setPosition($1);
-                        }
-    | LINKCOUNTED ROW {
-                            IHqlExpression* record = queryNullRecord();
-                            Owned<ITypeInfo> rowType = makeRowType(record->getType());
-                            $$.setType(setLinkCountedAttr(rowType, true));
-                            $$.setPosition($1);
-                        }
+    | explicitRowType
     | abstractModule
                         {
                             OwnedHqlExpr scope = $1.getExpr();

ecl/hql/hqlgram2.cpp

@@ -3624,7 +3624,7 @@ IHqlExpression* HqlGram::checkServiceDef(IHqlScope* serviceScope,IIdAtom * name,
             {
                 checkSvcAttrNoValue(attr, errpos);
             }
-            else if ((name == userMatchFunctionAtom) || (name == costAtom) || (name == allocatorAtom) || (name == extendAtom))
+            else if ((name == userMatchFunctionAtom) || (name == costAtom) || (name == allocatorAtom) || (name == extendAtom) || (name == passParameterMetaAtom))
             {
             }
             else if (name == holeAtom)

ecl/hqlcpp/hqlcpp.cpp

@@ -4279,25 +4279,8 @@ void HqlCppTranslator::buildTempExpr(BuildCtx & ctx, BuildCtx & declareCtx, CHql
     case type_row:
         {
             Owned<BoundRow> tempRow = declareTempRow(declareCtx, subctx, expr);
+            buildRowAssign(subctx, tempRow, expr);
             tempTarget.expr.set(tempRow->queryBound());
-            //MORE: This should be more general - i) to avoid unnecessary temporaries, and ii) to allow rows to be copied by linking.
-            if (isCall(expr->getOperator()))
-            {
-                buildExprAssign(subctx, tempTarget, expr);
-                return;
-            }
-
-            IHqlStmt * stmt = subctx.addGroup();
-            stmt->setIncomplete(true);
-
-            Owned<BoundRow> rowBuilder = createRowBuilder(subctx, tempRow);
-            Owned<IReferenceSelector> createdRef = createReferenceSelector(rowBuilder);
-            buildRowAssign(subctx, createdRef, expr);
-            finalizeTempRow(subctx, tempRow, rowBuilder);
-
-            stmt->setIncomplete(false);
-            stmt->mergeScopeWithContainer();
-            
             ctx.associate(*tempRow);
             break;
         }
@@ -5888,12 +5871,16 @@ void HqlCppTranslator::doBuildCall(BuildCtx & ctx, const CHqlBoundTarget * tgt,
         case type_table:
         case type_groupedtable:
             {
+                if (getBoolAttribute(external, passParameterMetaAtom, false))
+                    args.append(*buildMetaParameter(curParam));
                 ExpressionFormat format = queryNaturalFormat(argType);
                 buildDataset(ctx, castParam, bound, format);
                 break;
             }
         case type_row:
             {
+                if (getBoolAttribute(external, passParameterMetaAtom, false))
+                    args.append(*buildMetaParameter(curParam));
                 Owned<IReferenceSelector> selector = buildNewRow(ctx, castParam);
 
                 if (hasLinkCountedModifier(argType))

ecl/hqlcpp/hqlcpp.ipp

@@ -835,6 +835,7 @@ public:
     IReferenceSelector * buildActiveRow(BuildCtx & ctx, IHqlExpression * expr);
     IReferenceSelector * buildNewRow(BuildCtx & ctx, IHqlExpression * expr);
     IReferenceSelector * buildNewOrActiveRow(BuildCtx & ctx, IHqlExpression * expr, bool isNew);
+    void buildRowAssign(BuildCtx & ctx, BoundRow * target, IHqlExpression * expr);
     void buildRowAssign(BuildCtx & ctx, IReferenceSelector * target, IHqlExpression * expr);
     void buildRowAssign(BuildCtx & ctx, IReferenceSelector * target, IReferenceSelector * source);
     BoundRow * ensureLinkCountedRow(BuildCtx & ctx, BoundRow * row);
@@ -1076,6 +1077,7 @@ public:
     void finalizeTempRow(BuildCtx & ctx, BoundRow * targetRow, BoundRow * rowBuilder);
     BoundRow * declareTempAnonRow(BuildCtx & ctx, BuildCtx & codectx, IHqlExpression * record);
 
+    IHqlExpression * declareLinkedRowExpr(BuildCtx & ctx, IHqlExpression * record, bool isMember);
     BoundRow * declareLinkedRow(BuildCtx & ctx, IHqlExpression * expr, bool isMember);
     BoundRow * declareStaticRow(BuildCtx & ctx, IHqlExpression * expr);
 

ecl/hqlcpp/hqlcppds.cpp

@@ -261,11 +261,7 @@ IReferenceSelector * HqlCppTranslator::doBuildRowViaTemp(BuildCtx & ctx, IHqlExp
     }
 
     Owned<BoundRow> tempRow = declareTempRow(ctx, ctx, expr);
-
-    Owned<BoundRow> rowBuilder = createRowBuilder(ctx, tempRow);
-    Owned<IReferenceSelector> createdRef = createReferenceSelector(rowBuilder);
-    buildRowAssign(ctx, createdRef, expr);
-    finalizeTempRow(ctx, tempRow, rowBuilder);
+    buildRowAssign(ctx, tempRow, expr);
 
     ctx.associate(*tempRow);
     return createReferenceSelector(tempRow);
@@ -639,7 +635,8 @@ BoundRow * HqlCppTranslator::ensureLinkCountedRow(BuildCtx & ctx, BoundRow * row
         return row;
 
     OwnedHqlExpr srcRow = createTranslated(row->queryBound());
-    Owned<BoundRow> tempRow = declareLinkedRow(ctx, row->queryDataset(), false);
+    OwnedHqlExpr tempRowExpr = declareLinkedRowExpr(ctx, row->queryRecord(), false);
+    Owned<BoundRow> tempRow = row->clone(tempRowExpr);
 
     OwnedHqlExpr source = getPointer(row->queryBound());
     BuildCtx subctx(ctx);
@@ -660,7 +657,7 @@ BoundRow * HqlCppTranslator::ensureLinkCountedRow(BuildCtx & ctx, BoundRow * row
     generateExprCpp(s, boundSize.expr).append(",");
     generateExprCpp(s, source);
     s.append(")");
-    OwnedHqlExpr call = createQuoted(s, tempRow->queryBound()->queryType());
+    OwnedHqlExpr call = createQuoted(s, tempRow->queryBound()->getType());
 
     subctx.addAssign(tempRow->queryBound(), call);
 
@@ -4357,6 +4354,47 @@ void HqlCppTranslator::doBuildRowAssignUserTable(BuildCtx & ctx, IReferenceSelec
 }
 
 
+void HqlCppTranslator::buildRowAssign(BuildCtx & ctx, BoundRow * targetRow, IHqlExpression * expr)
+{
+    //MORE: We should improve assigning a link counted row to a dataset as well.
+    //The problem is that currently the dataset constructor is responsible for finializing the rows.
+    //which is more compact if the row can't just be appended.  Possibly needs an alwaysCreatesTemp()
+    //to help decide.
+    IHqlExpression * targetExpr = targetRow->queryBound();
+    if (targetRow->isLinkCounted() && hasWrapperModifier(targetExpr->queryType()))
+    {
+        CHqlBoundTarget target;
+        target.expr.set(targetRow->queryBound());
+
+        switch (expr->getOperator())
+        {
+        //MORE could support no_null, no_if, no_translated, constant no_createrow etc.
+        case no_call:
+        case no_externalcall:
+            buildExprAssign(ctx, target, expr);
+            return;
+        case no_comma:
+        case no_compound:
+            buildStmt(ctx, expr->queryChild(0));
+            buildRowAssign(ctx, targetRow, expr->queryChild(1));
+            return;
+        }
+    }
+
+    BuildCtx subctx(ctx);
+    IHqlStmt * stmt = subctx.addGroup();
+    stmt->setIncomplete(true);
+
+    Owned<BoundRow> rowBuilder = createRowBuilder(subctx, targetRow);
+    Owned<IReferenceSelector> createdRef = createReferenceSelector(rowBuilder);
+    buildRowAssign(subctx, createdRef, expr);
+    finalizeTempRow(subctx, targetRow, rowBuilder);
+
+    stmt->setIncomplete(false);
+    stmt->mergeScopeWithContainer();
+}
+
+
 void HqlCppTranslator::buildRowAssign(BuildCtx & ctx, IReferenceSelector * target, IHqlExpression * expr)
 {
     switch (expr->getOperator())

ecl/hqlcpp/hqlcppsys.ecl

@@ -844,7 +844,7 @@ const char * cppSystemText[]  = {
     "   bindBooleanParam(const varstring name, boolean val) : method,entrypoint='bindBooleanParam';",
     "   bindDataParam(const varstring name, data val) : method,entrypoint='bindDataParam';",
     "   bindRealParam(const varstring name, real val) : method,entrypoint='bindRealParam';",
-    "   bindRowParam(const varstring name, _linkcounted_ row row) : method,entrypoint='bindRowParam';",
+    "   bindRowParam(const varstring name, _linkcounted_ row row) : method,entrypoint='bindRowParam',passParameterMeta(true);",
     "   bindSignedParam(const varstring name, integer val) : method,entrypoint='bindSignedParam';",
     "   bindUnsignedParam(const varstring name, unsigned val) : method,entrypoint='bindUnsignedParam';",
     "   bindStringParam(const varstring name, const string val) : method,entrypoint='bindStringParam';",

ecl/hqlcpp/hqlhtcpp.cpp

@@ -1498,14 +1498,11 @@ BoundRow * HqlCppTranslator::createRowBuilder(BuildCtx & ctx, BoundRow * targetR
     }
 }
 
-BoundRow * HqlCppTranslator::declareLinkedRow(BuildCtx & ctx, IHqlExpression * expr, bool isMember)
+IHqlExpression * HqlCppTranslator::declareLinkedRowExpr(BuildCtx & ctx, IHqlExpression * record, bool isMember)
 {
-    assertex(expr->isDatarow());
-
     StringBuffer rowName;
     getUniqueId(rowName.append('r'));
 
-    IHqlExpression * record = expr->queryRecord();
     Owned<ITypeInfo> rowType = makeRowType(record->getType());
     rowType.setown(makeAttributeModifier(makeWrapperModifier(rowType.getClear()), getLinkCountedAttr()));
     if (isMember)
@@ -1518,6 +1515,13 @@ BoundRow * HqlCppTranslator::declareLinkedRow(BuildCtx & ctx, IHqlExpression * e
     //this the outer most scope (within a function)
     ctx.setNextPriority(BuildCtx::OutermostScopePrio);
     ctx.addDeclare(boundRow);
+    return boundRow.getClear();
+}
+
+BoundRow * HqlCppTranslator::declareLinkedRow(BuildCtx & ctx, IHqlExpression * expr, bool isMember)
+{
+    assertex(expr->isDatarow());
+    OwnedHqlExpr boundRow = declareLinkedRowExpr(ctx, expr->queryRecord(), isMember);
     return createBoundRow(expr, boundRow);
 }
 

ecl/hqlcpp/hqlwcpp.cpp

@@ -688,7 +688,7 @@ bool HqlCppWriter::generateFunctionPrototype(IHqlExpression * funcdef, const cha
             firstParam = false;
 
         IHqlExpression * param = formals->queryChild(i);
-        generateParamCpp(param);
+        generateParamCpp(param, body);
     }
     out.append(")");
     return true;
@@ -781,7 +781,7 @@ void HqlCppWriter::generateInitializer(IHqlExpression * expr)
 }
 
 
-void HqlCppWriter::generateParamCpp(IHqlExpression * param)
+void HqlCppWriter::generateParamCpp(IHqlExpression * param, IHqlExpression * attrs)
 {
     ITypeInfo *paramType = param->queryType();
     
@@ -808,6 +808,22 @@ void HqlCppWriter::generateParamCpp(IHqlExpression * param)
 
     switch (paramType->getTypeCode())
     {
+    case type_dictionary:
+    case type_table:
+    case type_groupedtable:
+    case type_row:
+        if (getBoolAttribute(attrs, passParameterMetaAtom, false))
+        {
+            out.append("IOutputMetaData & ");
+            if (paramName)
+                appendCapital(out.append(" meta"), paramNameText);
+            out.append(",");
+        }
+        break;
+    }
+
+    switch (paramType->getTypeCode())
+    {
     case type_string:
     case type_qstring:
     case type_data:
@@ -852,7 +868,7 @@ void HqlCppWriter::generateParamCpp(IHqlExpression * param)
     case type_row:
         isConst = true;
         break;
-    } 
+    }
     
     bool nameappended = false;
     switch (paramType->getTypeCode())
@@ -1359,7 +1375,7 @@ StringBuffer & HqlCppWriter::generateExprCpp(IHqlExpression * expr)
             generateExprCpp(expr->queryChild(0));
             break;
         case no_param:
-            generateParamCpp(expr);
+            generateParamCpp(expr, NULL);
             break;
         case no_callback:
             {

ecl/hqlcpp/hqlwcpp.ipp

@@ -85,7 +85,7 @@ protected:
     StringBuffer & generateChildExpr(StringBuffer & out, IHqlExpression * expr, unsigned childIndex);
     bool generateFunctionPrototype(IHqlExpression * funcdef, const char * name);
     void generateInitializer(IHqlExpression * expr);
-    void generateParamCpp(IHqlExpression * expr);
+    void generateParamCpp(IHqlExpression * expr, IHqlExpression * attrs);
     void generateSimpleAssign(IHqlExpression * target, IHqlExpression * source);
     void generateStmt(IHqlStmt * stmt);
     void generateStmtAssign(IHqlStmt * assign);

plugins/Rembed/Rembed.cpp

@@ -506,6 +506,14 @@ public:
             break;
         }
     }
+    virtual void bindRowParam(const char *name, IOutputMetaData & metaVal, byte *val)
+    {
+        UNIMPLEMENTED;
+    }
+    virtual void bindDatasetParam(const char *name, IOutputMetaData & metaVal, IRowStream * val)
+    {
+        UNIMPLEMENTED;
+    }
 
     virtual void importFunction(size32_t lenChars, const char *utf)
     {

plugins/javaembed/javaembed.cpp

@@ -1038,6 +1038,14 @@ public:
         argsig++;
         addArg(v);
     }
+    virtual void bindRowParam(const char *name, IOutputMetaData & metaVal, byte *val)
+    {
+        UNIMPLEMENTED;
+    }
+    virtual void bindDatasetParam(const char *name, IOutputMetaData & metaVal, IRowStream * val)
+    {
+        UNIMPLEMENTED;
+    }
 
     virtual void importFunction(size32_t lenChars, const char *utf)
     {

plugins/pyembed/pyembed.cpp

@@ -1012,6 +1012,14 @@ public:
         }
         addArg(name, vval.getLink());
     }
+    virtual void bindRowParam(const char *name, IOutputMetaData & metaVal, byte *val)
+    {
+        UNIMPLEMENTED;
+    }
+    virtual void bindDatasetParam(const char *name, IOutputMetaData & metaVal, IRowStream * val)
+    {
+        UNIMPLEMENTED;
+    }
 
 protected:
     virtual void addArg(const char *name, PyObject *arg) = 0;

plugins/v8embed/v8embed.cpp

@@ -246,6 +246,14 @@ public:
         }
         context->Global()->Set(v8::String::New(name), array);
     }
+    virtual void bindRowParam(const char *name, IOutputMetaData & metaVal, byte *val)
+    {
+        UNIMPLEMENTED;
+    }
+    virtual void bindDatasetParam(const char *name, IOutputMetaData & metaVal, IRowStream * val)
+    {
+        UNIMPLEMENTED;
+    }
 
     virtual bool getBooleanResult()
     {

rtl/eclrtl/eclrtl.hpp

@@ -801,6 +801,8 @@ interface IEmbedFunctionContext : extends IInterface
     virtual IRowStream *getDatasetResult(IEngineRowAllocator * _resultAllocator) = 0;
     virtual byte * getRowResult(IEngineRowAllocator * _resultAllocator) = 0;
     virtual size32_t getTransformResult(ARowBuilder & builder) = 0;
+    virtual void bindRowParam(const char *name, IOutputMetaData & metaVal, byte *val) = 0;
+    virtual void bindDatasetParam(const char *name, IOutputMetaData & metaVal, IRowStream * val) = 0;
 };
 
 interface IEmbedContext : extends IInterface

rtl/eclrtl/rtlds_imp.hpp

@@ -84,6 +84,7 @@ public:
     RtlFixedDatasetBuilder(unsigned _recordSize, unsigned _initRows);
 
     byte * createSelf();
+    virtual IEngineRowAllocator *queryAllocator() const { return NULL; }
 
 protected:
     unsigned recordSize;
@@ -112,6 +113,7 @@ public:
 
     void deserializeRow(IOutputRowDeserializer & deserializer, IRowDeserializerSource & in);
     virtual byte * createSelf();
+    virtual IEngineRowAllocator *queryAllocator() const { return NULL; }
 
 protected:
     IRecordSize * recordSize;

testing/ecl/streame2.ecl

@@ -71,4 +71,4 @@ output (testMissingTuple1(10));
 output (testMissingTuple2(10));
 
 output(testRowReturn(10));
-output(row(testTransform(10)));
+output(row(testTransform(10)));

testing/ecl/streame3.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;
+
+childrec := RECORD
+   string name => unsigned value;
+END;
+
+titleRec := { string title };
+titles := dataset(['', 'Mr. ', 'Rev. '], titleRec);
+
+// Test defining a transform
+transform(childrec) testTransformTitle(titleRec in, unsigned lim) := EMBED(Python)
+  return (in.title, lim)
+ENDEMBED;
+
+// Test defining a transform
+//MORE: The embed function shpo
+transform(childrec) testTransformTitle2(_linkcounted_ row(titleRec) in, unsigned lim) := EMBED(Python)
+  return (in.title, lim)
+ENDEMBED;
+
+sequential(
+output(project(titles, testTransformTitle(LEFT, 10)));
+output(project(titles, testTransformTitle2(LEFT, 10)));
+);