Merge branch 'candidate-5.2.0'

Signed-off-by: Richard Chapman <rchapman@hpccsystems.com>

Conflicts:
	esp/services/ws_ecl/ws_ecl_service.cpp
	esp/services/ws_ecl/ws_ecl_wuinfo.cpp

common/thorhelper/thorcommon.hpp

@@ -26,8 +26,15 @@
 #include "thorhelper.hpp"
 #include "thorxmlwrite.hpp"
 
-#define DALI_RESULT_OUTPUTMAX 2000 // MB
-#define DALI_RESULT_LIMIT_DEFAULT 10 // MB
+static unsigned const defaultDaliResultOutputMax = 2000; // MB
+static unsigned const defaultDaliResultLimit = 10; // MB
+static unsigned const defaultMaxCsvRowSize = 10; // MB
+
+
+#define OPT_OUTPUTLIMIT_LEGACY    "outputLimit"             // OUTPUT Mb limit (legacy property name, renamed to outputLimitMb in 5.2)
+#define OPT_OUTPUTLIMIT           "outputLimitMb"           // OUTPUT Mb limit                                                               (default = 10 [MB])
+#define OPT_MAXCSVROWSIZE         "maxCsvRowSizeMb"         // Upper limit on csv read line size                                             (default = 10 [MB])
+
 
 class THORHELPER_API CSizingSerializer : implements IRowSerializerTarget
 {

common/workunit/workunit.cpp

@@ -1624,12 +1624,14 @@ public:
     virtual IStringVal& getAttributeName(IStringVal &str) const;
     virtual IStringVal& getDefaultName(IStringVal &str) const;
     virtual IStringVal& getInfo(const char *name, IStringVal &str) const;
+    virtual IStringVal& getText(const char *name, IStringVal &str) const;
     virtual unsigned getWebServicesCRC() const;
  
     virtual void        setModuleName(const char *);
     virtual void        setAttributeName(const char *);
     virtual void        setDefaultName(const char *);
     virtual void        setInfo(const char *name, const char *info);
+    virtual void        setText(const char *name, const char *info);
     virtual void        setWebServicesCRC(unsigned);
 };
 
@@ -1701,6 +1703,8 @@ public:
     virtual unsigned    getResultHash() const;
     virtual bool        getResultIsAll() const;
     virtual IProperties *queryResultXmlns();
+    virtual IStringVal& getResultFieldOpt(const char *name, IStringVal &str) const;
+
 
     // interface IWUResult
     virtual void        setResultStatus(WUResultStatus status);
@@ -1734,6 +1738,7 @@ public:
     virtual void        setResultXML(const char *val);
     virtual void        setResultRow(unsigned len, const void * data);
     virtual void        setResultXmlns(const char *prefix, const char *uri);
+    virtual void        setResultFieldOpt(const char *name, const char *value);
 };
 
 class CLocalWUPlugin : public CInterface, implements IWUPlugin
@@ -4857,6 +4862,7 @@ void CLocalWorkUnit::copyWorkUnit(IConstWorkUnit *cached, bool all)
     copyTree(p, fromP, "Results");
     copyTree(p, fromP, "Graphs");
     copyTree(p, fromP, "Workflow");
+    copyTree(p, fromP, "WebServicesInfo");
     if (all)
     {
         // 'all' mode is used when setting up a dali WU from the embedded wu in a workunit dll
@@ -7069,6 +7075,11 @@ IStringVal& CLocalWUWebServicesInfo::getInfo(const char *name, IStringVal &str)
 
 }
 
+IStringVal& CLocalWUWebServicesInfo::getText(const char *name, IStringVal &str) const
+{
+    str.set(p->queryProp(name));
+    return str;
+}
 
 void CLocalWUWebServicesInfo::setModuleName(const char *mname)
 {
@@ -7098,6 +7109,11 @@ void CLocalWUWebServicesInfo::setInfo(const char *name, const char *info)
     p->setPropBin(name, m.length(), m.toByteArray());
 }
 
+void CLocalWUWebServicesInfo::setText(const char *name, const char *info)
+{
+    p->setProp(name, info);
+}
+
 
 //========================================================================================
 
@@ -7549,6 +7565,19 @@ IStringVal& CLocalWUResult::getResultFilename(IStringVal & str) const
     return str;
 }
 
+IStringVal& CLocalWUResult::getResultFieldOpt(const char *name, IStringVal &str) const
+{
+    str.clear();
+    if (!name || !*name)
+        return str;
+    IPropertyTree *format = p->queryPropTree("Format");
+    if (!format)
+        return str;
+    VStringBuffer xpath("@%s", name);
+    str.set(format->queryProp(xpath));
+    return str;
+}
+
 void CLocalWUResult::setResultStatus(WUResultStatus status)
 {
     setEnum(p, "@status", status, resultStatuses);
@@ -7575,6 +7604,16 @@ void CLocalWUResult::setResultXmlns(const char *prefix, const char *uri)
         xpath.append(':').append(prefix);
     p->setProp(xpath, uri);
 }
+
+void CLocalWUResult::setResultFieldOpt(const char *name, const char *value)
+{
+    if (!name || !*name)
+        return;
+    IPropertyTree *format = ensurePTree(p, "Format");
+    VStringBuffer xpath("@%s", name);
+    format->setProp(xpath, value);
+}
+
 void CLocalWUResult::setResultScalar(bool isScalar)
 {
     p->setPropInt("@isScalar", (int) isScalar);

common/workunit/workunit.hpp

@@ -312,6 +312,7 @@ interface IConstWUResult : extends IInterface
     virtual void getResultDecimal(void * val, unsigned length, unsigned precision, bool isSigned) const = 0;
     virtual bool getResultIsAll() const = 0;
     virtual const IProperties *queryResultXmlns() = 0;
+    virtual IStringVal &getResultFieldOpt(const char *name, IStringVal &str) const = 0;
 };
 
 
@@ -348,6 +349,7 @@ interface IWUResult : extends IConstWUResult
     virtual void setResultXML(const char * xml) = 0;
     virtual void setResultRow(unsigned len, const void * data) = 0;
     virtual void setResultXmlns(const char *prefix, const char *uri) = 0;
+    virtual void setResultFieldOpt(const char *name, const char *value)=0;
 };
 
 
@@ -426,6 +428,7 @@ interface IConstWUWebServicesInfo : extends IInterface
     virtual IStringVal & getDefaultName(IStringVal & str) const = 0;
     virtual IStringVal & getInfo(const char * name, IStringVal & str) const = 0;
     virtual unsigned getWebServicesCRC() const = 0;
+    virtual IStringVal & getText(const char * name, IStringVal & str) const = 0;
 };
 
 
@@ -436,6 +439,7 @@ interface IWUWebServicesInfo : extends IConstWUWebServicesInfo
     virtual void setDefaultName(const char * pstr) = 0;
     virtual void setInfo(const char * name, const char * info) = 0;
     virtual void setWebServicesCRC(unsigned crc) = 0;
+    virtual void setText(const char * name, const char * text) = 0;
 };
 
 

dali/base/dautils.cpp

@@ -138,8 +138,6 @@ public:
             return;
         StringArray lfnExpanded;
         StringBuffer tmp;
-        const char * s = prefix.str();
-        const char *start = strchr(s,'{');
         for (unsigned idx=0; idx < dlfns.size(); idx++)
         {
             const char *suffix = dlfns.at(idx).get();
@@ -152,7 +150,6 @@ public:
                     tmp.append(suffix);
                 tmp.clip().toLowerCase();
                 Owned<IDFAttributesIterator> iter=queryDistributedFileDirectory().getDFAttributesIterator(tmp.str(),_udesc,false,true,NULL);
-                prefix.setLength(start-s);
                 ForEach(*iter)
                 {
                     IPropertyTree &attr = iter->query();
@@ -326,7 +323,6 @@ void CDfsLogicalFileName::expand(IUserDescriptor *user)
             StringBuffer err;
             e->errorMessage(err);
             ERRLOG("CDfsLogicalFileName::expand %s",err.str());
-            e->Release();
             throw;
         }
     }

docs/ECLLanguageReference/ECLR_mods/BltInFunc-DENORMALIZE.xml

@@ -162,7 +162,7 @@ NamesTable := DATASET([ {0,'Kevin'},{0,'Liz'},{0,'Mr Nobody'},
 NormAddrs := DATASET([{'Kevin','10 Malt Lane'},
                       {'Liz','10 Malt Lane'},
                       {'Liz','3 The cottages'},
-                      {'Anywhee','Here'},
+                      {'Anywhere','Here'},
                       {'Anywhere','There'},
                       {'Anywhere','Near'},
                       {'Anywhere','Far'}],NormRec);

docs/RDDERef/RDDERef.xml

@@ -108,6 +108,11 @@
     the Roxie cluster by opening a socket to one of the servers in the
     cluster, or it can communicate via an ESP service such as WsECL.</para>
 
+    <para>Typically, Roxie results are returned to the requester rather than
+    writing a result to a file. However, Roxie can write data files, although
+    you generally would not want to write a file when a query is not
+    workunit-based. </para>
+
     <sect1 id="Roxie_Overview">
       <title>Roxie Overview</title>
 

ecl/eclcmd/eclcmd_common.cpp

@@ -380,6 +380,8 @@ public:
         cmdLine.set("eclcc -E");
         if (cmd.optLegacy)
             cmdLine.append(" -legacy");
+        if (cmd.optDebug)
+            cmdLine.append(" -g");
         appendOptPath(cmdLine, 'I', cmd.optImpPath.str());
         appendOptPath(cmdLine, 'L', cmd.optLibPath.str());
         if (cmd.optAttributePath.length())
@@ -586,6 +588,8 @@ eclCmdOptionMatchIndicator EclCmdWithEclTarget::matchCommandLineOption(ArgvItera
         return EclCmdOptionMatch;
     if (iter.matchFlag(optLegacy, ECLOPT_LEGACY) || iter.matchFlag(optLegacy, ECLOPT_LEGACY_DASH))
         return EclCmdOptionMatch;
+    if (iter.matchFlag(optDebug, ECLOPT_DEBUG) || iter.matchFlag(optDebug, ECLOPT_DEBUG_DASH))
+        return EclCmdOptionMatch;
     if (iter.matchOption(optResultLimit, ECLOPT_RESULT_LIMIT))
         return EclCmdOptionMatch;
     if (iter.matchOption(optTargetCluster, ECLOPT_CLUSTER_DEPRECATED)||iter.matchOption(optTargetCluster, ECLOPT_CLUSTER_DEPRECATED_S))

ecl/eclcmd/eclcmd_common.hpp

@@ -146,6 +146,8 @@ typedef IEclCommand *(*EclCommandFactory)(const char *cmdname);
 #define ECLOPT_MANIFEST_DASH "-manifest"
 #define ECLOPT_LEGACY "--legacy"
 #define ECLOPT_LEGACY_DASH "-legacy"
+#define ECLOPT_DEBUG "--debug"
+#define ECLOPT_DEBUG_DASH "-g"
 
 #define ECLOPT_VERBOSE "--verbose"
 #define ECLOPT_VERBOSE_S "-v"
@@ -243,7 +245,7 @@ public:
 class EclCmdWithEclTarget : public EclCmdCommon
 {
 public:
-    EclCmdWithEclTarget() : optLegacy(false), optNoArchive(false), optResultLimit((unsigned)-1)
+    EclCmdWithEclTarget() : optLegacy(false), optNoArchive(false), optResultLimit((unsigned)-1), optDebug(false)
     {
     }
     virtual eclCmdOptionMatchIndicator matchCommandLineOption(ArgvIterator &iter, bool finalAttempt=false);
@@ -264,6 +266,7 @@ public:
             "   -Lpath                 Add path to locations to search for system libraries\n"
             "   --manifest             Specify path to manifest file\n"
             "   --legacy               Use legacy import semantics (deprecated)\n"
+            "   --debug, -g            Enable debug symbols in generated code\n"
         );
     }
 public:
@@ -279,6 +282,7 @@ public:
     unsigned optResultLimit;
     bool optNoArchive;
     bool optLegacy;
+    bool optDebug;
 };
 
 class EclCmdWithQueryTarget : public EclCmdCommon

ecl/hql/hqlatoms.cpp

@@ -394,6 +394,7 @@ IAtom * stableAtom;
 IAtom * _state_Atom;
 IAtom * steppedAtom;
 IAtom * storedAtom;
+IAtom * storedFieldFormatAtom;
 IAtom * streamedAtom;
 IAtom * _streaming_Atom;
 IAtom * successAtom;
@@ -427,6 +428,7 @@ IAtom * virtualAtom;
 IAtom * _virtualSeq_Atom;
 IAtom * volatileAtom;
 IAtom * warningAtom;
+IAtom * webserviceAtom;
 IAtom * wholeAtom;
 IAtom * widthAtom;
 IAtom * wipeAtom;
@@ -823,6 +825,7 @@ MODULE_INIT(INIT_PRIORITY_HQLATOM)
     MAKESYSATOM(state);
     MAKEATOM(stepped);
     MAKEATOM(stored);
+    MAKEATOM(storedFieldFormat);
     MAKEATOM(streamed);
     MAKESYSATOM(streaming);
     MAKEATOM(success);
@@ -857,6 +860,7 @@ MODULE_INIT(INIT_PRIORITY_HQLATOM)
     MAKESYSATOM(virtualSeq);
     MAKEATOM(volatile);
     MAKEATOM(warning);
+    MAKEATOM(webservice);
     MAKEATOM(whole);
     MAKEATOM(width);
     MAKEATOM(wipe);

ecl/hql/hqlatoms.hpp

@@ -397,6 +397,7 @@ extern HQL_API IAtom * stableAtom;
 extern HQL_API IAtom * _state_Atom;
 extern HQL_API IAtom * steppedAtom;
 extern HQL_API IAtom * storedAtom;
+extern HQL_API IAtom * storedFieldFormatAtom;
 extern HQL_API IAtom * streamedAtom;
 extern HQL_API IAtom * _streaming_Atom;
 extern HQL_API IAtom * successAtom;
@@ -431,6 +432,7 @@ extern HQL_API IAtom * virtualAtom;
 extern HQL_API IAtom * _virtualSeq_Atom;
 extern HQL_API IAtom * volatileAtom;
 extern HQL_API IAtom * warningAtom;
+extern HQL_API IAtom * webserviceAtom;
 extern HQL_API IAtom * wholeAtom;
 extern HQL_API IAtom * widthAtom;
 extern HQL_API IAtom * wipeAtom;

ecl/hql/hqlgram.y

@@ -213,6 +213,7 @@ static void eclsyntaxerror(HqlGram * parser, const char * s, short yystate, int
   TOK_FIXED
   FLAT
   FROM
+  FORMAT
   FORMAT_ATTR
   FORWARD
   FROMJSON
@@ -552,6 +553,7 @@ static void eclsyntaxerror(HqlGram * parser, const char * s, short yystate, int
   HASH_STORED
   HASH_LINK
   HASH_ONWARNING
+  HASH_WEBSERVICE
 
   INTERNAL_READ_NEXT_TOKEN
 
@@ -1523,6 +1525,12 @@ setMetaCommand
                             }
                             $$.setExpr(createValue(no_setmeta, makeVoidType(), createAttribute(onWarningAtom), $3.getExpr(), $5.getExpr()), $1);
                         }
+    | HASH_WEBSERVICE '(' hintList ')'
+                        {
+                            HqlExprArray args;
+                            $3.unwindCommaList(args);
+                            $$.setExpr(createValue(no_setmeta, makeVoidType(), createExprAttribute(webserviceAtom, args)), $1);
+                        }
     ;
 
 hashStoredValue
@@ -1604,7 +1612,12 @@ failure
                             parser->normalizeExpression($5, type_string, true);
                             $$.setExpr(createValueF(no_persist, makeVoidType(), $3.getExpr(), $5.getExpr(), $6.getExpr(), NULL), $1);
                         }
-    | STORED '(' expression optFewMany ')'
+    | STORED '(' expression ',' fewMany optStoredFieldFormat ')'
+                        {
+                            parser->normalizeStoredNameExpression($3);
+                            $$.setExpr(createValue(no_stored, makeVoidType(), $3.getExpr(), $5.getExpr(), $6.getExpr()), $1);
+                        }
+    | STORED '(' expression optStoredFieldFormat ')'
                         {
                             parser->normalizeStoredNameExpression($3);
                             $$.setExpr(createValue(no_stored, makeVoidType(), $3.getExpr(), $4.getExpr()), $1);
@@ -1715,6 +1728,18 @@ persistOpt
                         }
     ;
 
+optStoredFieldFormat
+    :                   {
+                            $$.setNullExpr();
+                        }
+    | ',' FORMAT '(' hintList ')'
+                        {
+                            HqlExprArray args;
+                            $4.unwindCommaList(args);
+                            $$.setExpr(createExprAttribute(storedFieldFormatAtom, args), $2);
+                        }
+    ;
+
 globalOpts
     :                   {   $$.setNullExpr(); }
     | ',' globalOpts2   {   $$.inherit($2); }

ecl/hql/hqlgram2.cpp

@@ -10466,6 +10466,7 @@ static void getTokenText(StringBuffer & msg, int token)
     case FIRST: msg.append("FIRST"); break;
     case TOK_FIXED: msg.append("FIXED"); break;
     case FLAT: msg.append("FLAT"); break;
+    case FORMAT: msg.append("FORMAT"); break;
     case FORMAT_ATTR: msg.append("FORMAT"); break;
     case FORWARD: msg.append("FORWARD"); break;
     case FROM: msg.append("FROM"); break;
@@ -10731,6 +10732,7 @@ static void getTokenText(StringBuffer & msg, int token)
     case HASH_STORED: msg.append("#STORED"); break;
     case HASH_LINK: msg.append("#LINK"); break;
     case HASH_WORKUNIT: msg.append("#WORKUNIT"); break;
+    case HASH_WEBSERVICE: msg.append("#WEBSERVICE"); break;
     case SIMPLE_TYPE: msg.append("type-name"); break;
 
     case EQ: msg.append("="); break;

ecl/hql/hqllex.l

@@ -565,6 +565,12 @@ xpathseq      ([^}\r\n])+
                             return SKIPPED; 
                         return(HASH_WORKUNIT); 
                     }
+#WEBSERVICE         {
+                        setupdatepos;
+                        if (lexer->skipNesting || lexer->macroGathering)
+                            return SKIPPED;
+                        return(HASH_WEBSERVICE);
+                    }
 __LINE__            {
                         setupdatepos; 
                         returnToken.setExpr(createIntegerConstant(lexer->yyLineNo, false));
@@ -699,6 +705,7 @@ FILTERED            { RETURNSYM(FILTERED); }
 FIRST               { RETURNSYM(FIRST); }
 FIXED               { RETURNSYM(TOK_FIXED); }
 FLAT                { RETURNSYM(FLAT); }
+FORMAT              { RETURNSYM(FORMAT); }
 FROM                { RETURNSYM(FROM); }
 FORMAT              { RETURNSYM(FORMAT_ATTR); }
 FORWARD             { RETURNSYM(FORWARD); }

ecl/hql/hqlthql.cpp

@@ -2368,6 +2368,8 @@ void HqltHql::toECL(IHqlExpression *expr, StringBuffer &s, bool paren, bool inTy
                     s.append("#STORED");
                 else if (kind == workunitAtom)
                     s.append("#WORKUNIT");
+                else if (kind == webserviceAtom)
+                    s.append("#WEBSERVICE");
                 else
                     s.append("#META:").append(kind);
 

ecl/hql/hqlutil.cpp

@@ -2994,6 +2994,64 @@ bool isValidXmlRecord(IHqlExpression * expr)
     return true;
 }
 
+static void expandHintValue(StringBuffer & s, IHqlExpression * expr)
+{
+    node_operator op = expr->getOperator();
+    node_operator childOp = no_none;
+    switch (op)
+    {
+    case no_constant:
+        expr->queryValue()->getStringValue(s);
+        break;
+    case no_comma:
+        expandHintValue(s, expr->queryChild(0));
+        expandHintValue(s.append(","), expr->queryChild(1));
+        break;
+    case no_range:
+        expandHintValue(s, expr->queryChild(0));
+        expandHintValue(s.append(".."), expr->queryChild(1));
+        break;
+    case no_rangefrom:
+        expandHintValue(s, expr->queryChild(0));
+        s.append("..");
+        break;
+    case no_rangeto:
+        expandHintValue(s.append(".."), expr->queryChild(0));
+        break;
+    case no_list:
+        {
+            s.append("[");
+            ForEachChild(i, expr)
+            {
+                if (i)
+                    s.append(",");
+                expandHintValue(s, expr->queryChild(i));
+            }
+            s.append("]");
+            break;
+        }
+    case no_attr:
+        s.append(expr->queryName());
+        break;
+    default:
+        s.append("?");
+        break;
+    }
+}
+
+void getHintNameValue(IHqlExpression * attr, StringBuffer &name, StringBuffer &value)
+{
+    name.set(attr->queryName()->str());
+    ForEachChild(i, attr)
+    {
+        if (i)
+            value.append(",");
+        expandHintValue(value, attr->queryChild(i));
+    }
+    if (value.length() == 0)
+        value.append("1");
+}
+
 bool getBoolValue(IHqlExpression * expr, bool dft)
 {
     if (expr)

ecl/hql/hqlutil.hpp

@@ -111,6 +111,7 @@ extern HQL_API bool isValidXmlRecord(IHqlExpression * expr);
 extern HQL_API bool matchesConstantValue(IHqlExpression * expr, __int64 test);
 extern HQL_API bool matchesBoolean(IHqlExpression * expr, bool test);
 extern HQL_API bool matchesConstantString(IHqlExpression * expr, const char * text, bool ignoreCase);
+extern HQL_API void getHintNameValue(IHqlExpression * attr, StringBuffer &name, StringBuffer &value);
 extern HQL_API bool getBoolValue(IHqlExpression * expr, bool dft);
 extern HQL_API __int64 getIntValue(IHqlExpression * expr, __int64 dft = 0);
 extern HQL_API StringBuffer & getStringValue(StringBuffer & out, IHqlExpression * expr, const char * dft = NULL);

ecl/hqlcpp/hqlhtcpp.cpp

@@ -1880,49 +1880,6 @@ void ActivityInstance::removeAttribute(const char * name)
     removeGraphAttribute(graphNode, name);
 }
 
-static void expandHintValue(StringBuffer & s, IHqlExpression * expr)
-{
-    switch (expr->getOperator())
-    {
-    case no_constant:
-        expr->queryValue()->getStringValue(s);
-        break;
-    case no_comma:
-        expandHintValue(s, expr->queryChild(0));
-        expandHintValue(s.append(","), expr->queryChild(1));
-        break;
-    case no_range:
-        expandHintValue(s, expr->queryChild(0));
-        expandHintValue(s.append(".."), expr->queryChild(1));
-        break;
-    case no_rangefrom:
-        expandHintValue(s, expr->queryChild(0));
-        s.append("..");
-        break;
-    case no_rangeto:
-        expandHintValue(s.append(".."), expr->queryChild(0));
-        break;
-    case no_list:
-        {
-            s.append("[");
-            ForEachChild(i, expr)
-            {
-                if (i)
-                    s.append(",");
-                expandHintValue(s, expr->queryChild(i));
-            }
-            s.append("]");
-            break;
-        }
-    case no_attr:
-        s.append(expr->queryName());
-        break;
-    default:
-        s.append("?");
-        break;
-    }
-}
-
 void ActivityInstance::processAnnotation(IHqlExpression * annotation)
 {
     switch (annotation->getAnnotationKind())
@@ -1966,20 +1923,12 @@ void ActivityInstance::processAnnotations(IHqlExpression * expr)
 
 void ActivityInstance::processHint(IHqlExpression * attr)
 {
-    IAtom * name = attr->queryName();
+    StringBuffer name;
     StringBuffer value;
-    ForEachChild(i, attr)
-    {
-        if (i)
-            value.append(",");
-        expandHintValue(value, attr->queryChild(i));
-    }
-    if (value.length() == 0)
-        value.append("1");
-
+    getHintNameValue(attr, name, value);
     IPropertyTree * att = createPTree();
-    att->setProp("@name", name->str());
-    att->setProp("@value", value.str());
+    att->setProp("@name", name);
+    att->setProp("@value", value);
     graphNode->addPropTree("hint", att);
 }
 
@@ -5299,12 +5248,13 @@ void HqlCppTranslator::buildSetResultInfo(BuildCtx & ctx, IHqlExpression * origi
     {
         HqlExprArray xmlnsAttrs;
         gatherAttributes(xmlnsAttrs, xmlnsAtom, originalExpr);
+        Owned<IWUResult> result;
         if (retType == type_row)
         {
             OwnedHqlExpr record = LINK(::queryRecord(schemaType));
             if (originalExpr->hasAttribute(noXpathAtom))
                 record.setown(removeAttributeFromFields(record, xpathAtom));
-            Owned<IWUResult> result = createDatasetResultSchema(seq, name, record, xmlnsAttrs, false, false);
+            result.setown(createDatasetResultSchema(seq, name, record, xmlnsAttrs, false, false));
             if (result)
                 result->setResultTotalRowCount(1);
         }
@@ -5312,7 +5262,7 @@ void HqlCppTranslator::buildSetResultInfo(BuildCtx & ctx, IHqlExpression * origi
         {
             // Bit of a mess - should split into two procedures
             int sequence = (int) seq->queryValue()->getIntValue();
-            Owned<IWUResult> result = createWorkunitResult(sequence, name);
+            result.setown(createWorkunitResult(sequence, name));
             if(result)
             {
                 StringBuffer fieldName;
@@ -5346,6 +5296,19 @@ void HqlCppTranslator::buildSetResultInfo(BuildCtx & ctx, IHqlExpression * origi
                 addSchemaResource(sequence, resultName.str(), xml.length()+1, xml.str());
             }
         }
+
+        IHqlExpression * format =  originalExpr->queryAttribute(storedFieldFormatAtom);
+        if (format)
+        {
+            ForEachChild(i, format)
+            {
+                StringBuffer name;
+                StringBuffer value;
+                OwnedHqlExpr folded = foldHqlExpression(format->queryChild(i));
+                getHintNameValue(folded, name, value);
+                result->setResultFieldOpt(name, value);
+            }
+        }
     }
 }
 

ecl/hqlcpp/hqlttcpp.cpp

@@ -164,6 +164,7 @@ public:
     OwnedHqlExpr storedName;
     OwnedHqlExpr originalLabel;
     OwnedHqlExpr sequence;
+    OwnedHqlExpr fieldFormat;
     node_operator setOp;
     node_operator persistOp;
 protected:
@@ -281,7 +282,19 @@ void NewThorStoredReplacer::doAnalyseBody(IHqlExpression * expr)
         StringBuffer errorTemp;
         seenMeta = true;
         IAtom * kind = expr->queryChild(0)->queryName();
-        if (kind == debugAtom)
+        if (kind == webserviceAtom)
+        {
+            Owned<IWUWebServicesInfo> wsi = wu->updateWebServicesInfo(true);
+            IHqlExpression *wsExpr = expr->queryChild(0);
+            ForEachChild(i, wsExpr)
+            {
+                StringBuffer name, value;
+                OwnedHqlExpr folded = foldHqlExpression(wsExpr->queryChild(i));
+                getHintNameValue(folded, name, value);
+                wsi->setText(name, value);
+            }
+        }
+        else if (kind == debugAtom)
         {
             OwnedHqlExpr foldedName = foldHqlExpression(expr->queryChild(1));
             OwnedHqlExpr foldedValue = foldHqlExpression(expr->queryChild(2));
@@ -5016,6 +5029,8 @@ IHqlExpression * GlobalAttributeInfo::createSetValue(IHqlExpression * value, IHq
         extraSetAttr->unwindList(args, no_comma);
     if (cluster)
         args.append(*createAttribute(clusterAtom, LINK(cluster)));
+    if (fieldFormat)
+        args.append(*LINK(fieldFormat));
     if (setOp == no_setresult)
         return createSetResult(args);
     return createValue(setOp, makeVoidType(), args);
@@ -5058,6 +5073,7 @@ void GlobalAttributeInfo::extractStoredInfo(IHqlExpression * expr, IHqlExpressio
         storedName.set(expr->queryChild(0));
         originalLabel.set(storedName);
         sequence.setown(getStoredSequenceNumber());
+        fieldFormat.set(expr->queryAttribute(storedFieldFormatAtom));
         few = true;
         break;
     case no_checkpoint:

ecl/hthor/hthor.cpp

@@ -5944,11 +5944,12 @@ void CHThorWorkUnitWriteActivity::execute()
 {
     unsigned flags = helper.getFlags();
     grouped = (POFgrouped & flags) != 0;
-    size32_t outputLimit = agent.queryWorkUnit()->getDebugValueInt("outputLimit", DALI_RESULT_LIMIT_DEFAULT);
+    // In absense of OPT_OUTPUTLIMIT check pre 5.2 legacy name OPT_OUTPUTLIMIT_LEGACY
+    size32_t outputLimit = agent.queryWorkUnit()->getDebugValueInt(OPT_OUTPUTLIMIT, agent.queryWorkUnit()->getDebugValueInt(OPT_OUTPUTLIMIT_LEGACY, defaultDaliResultLimit));
     if (flags & POFmaxsize)
         outputLimit = helper.getMaxSize();
-    if (outputLimit>DALI_RESULT_OUTPUTMAX)
-        throw MakeStringException(0, "Dali result outputs are restricted to a maximum of %d MB, the current limit is %d MB. A huge dali result usually indicates the ECL needs altering.", DALI_RESULT_OUTPUTMAX, DALI_RESULT_LIMIT_DEFAULT);
+    if (outputLimit>defaultDaliResultOutputMax)
+        throw MakeStringException(0, "Dali result outputs are restricted to a maximum of %d MB, the current limit is %d MB. A huge dali result usually indicates the ECL needs altering.", defaultDaliResultOutputMax, defaultDaliResultLimit);
     assertex(outputLimit<=0x1000); // 32bit limit because MemoryBuffer/CMessageBuffers involved etc.
     outputLimit *= 0x100000;
     MemoryBuffer rowdata;
@@ -6089,7 +6090,8 @@ void CHThorDictionaryWorkUnitWriteActivity::execute()
     }
     size32_t usedCount = rtlDictionaryCount(builder.getcount(), builder.queryrows());
 
-    size32_t outputLimit = agent.queryWorkUnit()->getDebugValueInt("outputLimit", DALI_RESULT_LIMIT_DEFAULT) * 0x100000;
+    // In absense of OPT_OUTPUTLIMIT check pre 5.2 legacy name OPT_OUTPUTLIMIT_LEGACY
+    size32_t outputLimit = agent.queryWorkUnit()->getDebugValueInt(OPT_OUTPUTLIMIT, agent.queryWorkUnit()->getDebugValueInt(OPT_OUTPUTLIMIT_LEGACY, defaultDaliResultLimit)) * 0x100000;
     MemoryBuffer rowdata;
     CThorDemoRowSerializer out(rowdata);
     Owned<IOutputRowSerializer> serializer = input->queryOutputMeta()->createDiskSerializer(agent.queryCodeContext(), activityId);
@@ -8717,6 +8719,7 @@ const void *CHThorDiskGroupAggregateActivity::nextInGroup()
 
 CHThorCsvReadActivity::CHThorCsvReadActivity(IAgentContext &_agent, unsigned _activityId, unsigned _subgraphId, IHThorCsvReadArg &_arg, ThorActivityKind _kind) : CHThorDiskReadBaseActivity(_agent, _activityId, _subgraphId, _arg, _kind), helper(_arg)
 {
+    maxRowSize = agent.queryWorkUnit()->getDebugValueInt(OPT_MAXCSVROWSIZE, defaultMaxCsvRowSize) * 1024 * 1024;
 }
 
 CHThorCsvReadActivity::~CHThorCsvReadActivity()
@@ -8778,7 +8781,6 @@ const void *CHThorCsvReadActivity::nextInGroup()
         if (!inputstream->eos())
         {
             size32_t rowSize = 4096; // MORE - make configurable
-            size32_t maxRowSize = 10*1024*1024; // MORE - make configurable
             size32_t thisLineLength;
             loop
             {

ecl/hthor/hthor.ipp

@@ -2299,6 +2299,7 @@ protected:
     CSVSplitter         csvSplitter;    
     unsigned __int64 limit;
     unsigned __int64 stopAfter;
+    size32_t maxRowSize;
 };
 
 class CHThorXmlReadActivity : public CHThorDiskReadBaseActivity, implements IXMLSelect

esp/services/ws_ecl/ws_ecl_service.cpp

@@ -1119,14 +1119,12 @@ void appendEclInputXsds(StringBuffer &content, IPropertyTree *xsd, BoolHash &add
     }
 }
 
-void CWsEclBinding::SOAPSectionToXsd(WsEclWuInfo &wsinfo, const char *parmXml, StringBuffer &schema, bool isRequest, IPropertyTree *xsdtree)
+void CWsEclBinding::SOAPSectionToXsd(WsEclWuInfo &wuinfo, IPropertyTree *parmTree, StringBuffer &schema, bool isRequest, IPropertyTree *xsdtree)
 {
-    Owned<IPropertyTree> tree = createPTreeFromXMLString(parmXml);
-
-    schema.appendf("<xsd:element name=\"%s%s\">", wsinfo.queryname.str(), isRequest ? "Request" : "Response");
+    schema.appendf("<xsd:element name=\"%s%s\">", wuinfo.queryname.str(), isRequest ? "Request" : "Response");
     schema.append("<xsd:complexType>");
     schema.append("<xsd:all>");
-    Owned<IPropertyTreeIterator> parts = tree->getElements("part");
+    Owned<IPropertyTreeIterator> parts = parmTree->getElements("part");
     if (parts)
     {
         ForEach(*parts)
@@ -1151,13 +1149,16 @@ void CWsEclBinding::SOAPSectionToXsd(WsEclWuInfo &wsinfo, const char *parmXml, S
             }
 
             schema.appendf("<xsd:element minOccurs=\"0\" maxOccurs=\"1\" name=\"%s\" type=\"%s\"", name, type.str());
-            if (strieq(type.str(), "tns:XmlDataSet"))
+            if (part.hasProp("@width") || part.hasProp("@height"))
             {
-                schema.append(">"
-                        "<xsd:annotation><xsd:appinfo>"
-                            "<form formRows=\"25\" formCols=\"60\"/>"
-                        "</xsd:appinfo></xsd:annotation>"
-                    "</xsd:element>");
+                schema.append("><xsd:annotation><xsd:appinfo><form");
+                unsigned rows = part.getPropInt("@height");
+                if (rows)
+                    schema.appendf(" formRows='%u'", rows);
+                unsigned cols = part.getPropInt("@width");
+                if (cols)
+                    schema.appendf(" formCols='%u'", cols);
+                schema.appendf("/></xsd:appinfo></xsd:annotation></xsd:element>");
             }
             else
                 schema.append("/>");
@@ -1213,20 +1214,21 @@ int CWsEclBinding::getXsdDefinition(IEspContext &context, CHttpRequest *request,
 
         if (wsinfo.queryname.length()>0)
         {
-            StringBuffer parmXml;
-            if (wsinfo.getWsResource("SOAP", parmXml))
+
+            IPropertyTree *parmTree = wsinfo.queryParamInfo();
+            if (xsdtree)
             {
-                if (xsdtree)
+                BoolHash added;
+                Owned<IPropertyTreeIterator> input_xsds =xsdtree->getElements("Input");
+                ForEach (*input_xsds)
                 {
-                    BoolHash added;
-                    Owned<IPropertyTreeIterator> input_xsds =xsdtree->getElements("Input");
-                    ForEach (*input_xsds)
-                    {
-                        appendEclInputXsds(content, &input_xsds->query(), added);
-                    }
+                    IPropertyTree &input = input_xsds->query();
+                    VStringBuffer xpath("part[@name='%s']", input.queryProp("@name"));
+                    if (parmTree->hasProp(xpath))
+                        appendEclInputXsds(content, &input, added);
                 }
-                SOAPSectionToXsd(wsinfo, parmXml.str(), content, true, xsdtree);
             }
+            SOAPSectionToXsd(wsinfo, parmTree, content, true, xsdtree);
 
             content.appendf("<xsd:element name=\"%sResponse\">", wsinfo.queryname.str());
             content.append("<xsd:complexType>");
@@ -1323,9 +1325,9 @@ bool CWsEclBinding::getSchema(StringBuffer& schema, IEspContext &ctx, CHttpReque
     return true;
 }
 
-int CWsEclBinding::getGenForm(IEspContext &context, CHttpRequest* request, CHttpResponse* response, WsEclWuInfo &wsinfo, bool box)
+int CWsEclBinding::getGenForm(IEspContext &context, CHttpRequest* request, CHttpResponse* response, WsEclWuInfo &wuinfo, bool box)
 {
-    IConstWorkUnit *wu = wsinfo.ensureWorkUnit();
+    IConstWorkUnit *wu = wuinfo.ensureWorkUnit();
     IProperties *parms = request->queryParameters();
 
     StringBuffer page;
@@ -1333,24 +1335,43 @@ int CWsEclBinding::getGenForm(IEspContext &context, CHttpRequest* request, CHttp
 
     StringBuffer v;
     StringBuffer formxml("<FormInfo>");
-    appendXMLTag(formxml, "WUID", wsinfo.queryWuid());
-    appendXMLTag(formxml, "QuerySet", wsinfo.qsetname.str());
-    appendXMLTag(formxml, "QueryName", wsinfo.queryname.str());
+    appendXMLTag(formxml, "WUID", wuinfo.queryWuid());
+    appendXMLTag(formxml, "QuerySet", wuinfo.qsetname.str());
+    appendXMLTag(formxml, "QueryName", wuinfo.queryname.str());
     appendXMLTag(formxml, "ClientVersion", v.appendf("%g",context.getClientVersion()).str());
-    appendXMLTag(formxml, "RequestElement", v.clear().append(wsinfo.queryname).append("Request").str());
+    appendXMLTag(formxml, "RequestElement", v.clear().append(wuinfo.queryname).append("Request").str());
+
+    StringBuffer help;
+    StringBuffer info;
+
+    Owned<IConstWUWebServicesInfo> ws = wu->getWebServicesInfo();
+    if (ws)
+    {
+        StringBufferAdaptor helpSv(help);
+        StringBufferAdaptor infoSv(info);
+
+        ws->getText("help", helpSv);
+        ws->getText("description", infoSv);
+    }
 
-    Owned<IWuWebView> web = createWuWebView(*wu, wsinfo.qsetname.get(), wsinfo.queryname.get(), getCFD(), true);
+    Owned<IWuWebView> web = createWuWebView(*wu, wuinfo.qsetname.get(), wuinfo.queryname.get(), getCFD(), true);
     if (web)
     {
-        appendXMLTag(formxml, "Help", web->aggregateResources("HELP", v.clear()).str());
-        appendXMLTag(formxml, "Info", web->aggregateResources("INFO", v.clear()).str());
+        if (!help.length())
+            web->aggregateResources("HELP", help);
+        if (!info.length())
+            web->aggregateResources("INFO", info);;
     }
+    if (help.length())
+        appendXMLTag(formxml, "Help", help.str());
+    if (info.length())
+        appendXMLTag(formxml, "Info", info.str());
 
     context.addOptions(ESPCTX_ALL_ANNOTATION);
     if (box)
     {
         StringBuffer xmlreq;
-        getWsEcl2XmlRequest(xmlreq, context, request, wsinfo, "xml", NULL, 0, true);
+        getWsEcl2XmlRequest(xmlreq, context, request, wuinfo, "xml", NULL, 0, true);
         if (xmlreq.length())
         {
             Owned<IPropertyTree> pretty = createPTreeFromXMLString(xmlreq.str(), ipt_ordered);
@@ -1364,7 +1385,7 @@ int CWsEclBinding::getGenForm(IEspContext &context, CHttpRequest* request, CHttp
         }
     }
     else
-        getSchema(formxml, context, request, wsinfo);
+        getSchema(formxml, context, request, wuinfo);
 
     formxml.append("<CustomViews>");
     if (web)

esp/services/ws_ecl/ws_ecl_service.hpp

@@ -161,7 +161,7 @@ public:
     void appendSchemaNamespaces(IPropertyTree *namespaces, IEspContext &ctx, CHttpRequest* req, WsEclWuInfo &wsinfo);
     void appendSchemaNamespaces(IPropertyTree *namespaces, IEspContext &ctx, CHttpRequest* req, const char *service, const char *method);
 
-    void SOAPSectionToXsd(WsEclWuInfo &wsinfo, const char *parmXml, StringBuffer &schema, bool isRequest=true, IPropertyTree *xsdtree=NULL);
+    void SOAPSectionToXsd(WsEclWuInfo &wsinfo, IPropertyTree *parmTree, StringBuffer &schema, bool isRequest=true, IPropertyTree *xsdtree=NULL);
     int getXmlTestForm(IEspContext &context, CHttpRequest* request, CHttpResponse* response, const char *formtype, WsEclWuInfo &wsinfo);
     int getXmlTestForm(IEspContext &context, CHttpRequest* request, CHttpResponse* response, WsEclWuInfo &wsinfo, const char *formtype);
 

esp/services/ws_ecl/ws_ecl_wuinfo.cpp

@@ -45,74 +45,139 @@ IConstWorkUnit *WsEclWuInfo::ensureWorkUnit()
     return wu;
 }
 
+void appendVariableParmInfo(IArrayOf<IPropertyTree> &parts, IResultSetFactory *resultSetFactory, IConstWUResult &var, unsigned hashWebserviceSeq=0)
+{
+    SCMStringBuffer varname;
+    var.getResultName(varname);
+    int seq = var.getResultSequence();
+
+    WUResultFormat fmt = var.getResultFormat();
+
+    SCMStringBuffer eclschema;
+    var.getResultEclSchema(eclschema);
+
+    SCMStringBuffer s;
+    Owned<IResultSetMetaData> meta = resultSetFactory->createResultSetMeta(&var);
+    StringBuffer width, height, fieldSeq;
+    var.getResultFieldOpt("fieldwidth", StringBufferAdaptor(width));
+    var.getResultFieldOpt("fieldheight", StringBufferAdaptor(height));
+    if (hashWebserviceSeq)
+        fieldSeq.append(hashWebserviceSeq);
+    else
+        var.getResultFieldOpt("sequence", StringBufferAdaptor(fieldSeq));
+
+    Owned<IPropertyTree> part = createPTree("part");
+    if (!var.isResultScalar())
+    {
+        meta->getXmlSchema(s, false);
+        part->setProp("@name", varname.str());
+        part->setProp("@type", "tns:XmlDataset");
+        if (fieldSeq.length())
+            part->setProp("@sequence", fieldSeq);
+    }
+    else
+    {
+        meta->getColumnEclType(s, 0);
+        DisplayType dt = meta->getColumnDisplayType(0);
+        StringAttr ptype;
+        switch (dt)
+        {
+        case TypeBoolean:
+            ptype.set("xsd:boolean");
+            break;
+        case TypeInteger:
+            ptype.set("xsd:integer");
+            break;
+        case TypeUnsignedInteger:
+            ptype.set("xsd:integer");
+            break;
+        case TypeReal:
+            ptype.set("xsd:real");
+            break;
+        case TypeSet:
+            ptype.set("tns:EspStringArray");
+            break;
+        case TypeDataset:
+        case TypeData:
+            ptype.set("tns:XmlDataSet");
+            break;
+        case TypeUnicode:
+        case TypeString:
+            ptype.set("xsd:string");
+            break;
+        case TypeUnknown:
+        case TypeBeginIfBlock:
+        case TypeEndIfBlock:
+        case TypeBeginRecord:
+        default:
+            ptype.set("xsd:string");
+            break;
+        }
+        part->setProp("@name", varname.str());
+        part->setProp("@type", ptype.str());
+        if (width.length())
+            part->setProp("@width", width);
+        if (height.length())
+            part->setProp("@height", height);
+        if (fieldSeq.length())
+            part->setProp("@sequence", fieldSeq);
+    }
+    parts.append(*part.getClear());
+}
+
+int orderParts(IInterface * const * pLeft, IInterface * const * pRight)
+{
+    IPropertyTree * right = (IPropertyTree *)*pRight;
+    IPropertyTree * left = (IPropertyTree *)*pLeft;
+    bool hasRightSeq = right->hasProp("@sequence");
+    bool hasLeftSeq = left->hasProp("@sequence");
+    if (hasRightSeq && hasLeftSeq)
+        return left->getPropInt("@sequence") - right->getPropInt("@sequence");
+    if (hasRightSeq);
+        return -1;
+    if (hasLeftSeq)
+        return 1;
+    if (!right->hasProp("@name"))
+        return -1;
+    if (!left->hasProp("@name"))
+        return 1;
+    return stricmp(right->queryProp("@name"), left->queryProp("@name"));  //fields without sequence alphabetical AFTER sequenced fields
+}
+
 bool WsEclWuInfo::getWsResource(const char *name, StringBuffer &out)
 {
     if (strieq(name, "SOAP"))
     {
         out.appendf("<message name=\"%s\">", queryname.str());
-        IConstWUResultIterator &vars = ensureWorkUnit()->getVariables();
-        Owned<IResultSetFactory> resultSetFactory(getResultSetFactory(username, password));
-        ForEach(vars)
+        Owned<IResultSetFactory> resultSetFactory = getResultSetFactory(username, password);
+        Owned<IConstWUWebServicesInfo> wsinfo = ensureWorkUnit()->getWebServicesInfo();
+        StringArray fields;
+        if (wsinfo)
         {
-            IConstWUResult &var = vars.query();
-            SCMStringBuffer varname;
-            var.getResultName(varname);
-            int seq = var.getResultSequence();
-
-            WUResultFormat fmt = var.getResultFormat();
-
-            SCMStringBuffer eclschema;
-            var.getResultEclSchema(eclschema);
-
-            SCMStringBuffer s;
-            Owned<IResultSetMetaData> meta = resultSetFactory->createResultSetMeta(&var);
-
-            if (!var.isResultScalar())
-            {
-                meta->getXmlSchema(s, false);
-                out.appendf("<part name=\"%s\" type=\"tns:XmlDataSet\" />", varname.str());
-            }
-            else
+            SCMStringBuffer fieldList;
+            wsinfo->getText("fields", fieldList);
+            if (fieldList.length())
+                fields.appendListUniq(fieldList.str(), ",");
+        }
+        IArrayOf<IPropertyTree> parts;
+        if (fields.length())
+        {
+            ForEachItemIn(i, fields)
             {
-                meta->getColumnEclType(s, 0);
-                DisplayType dt = meta->getColumnDisplayType(0);
-                StringAttr ptype;
-                switch (dt)
-                {
-                case TypeBoolean:
-                    ptype.set("xsd:boolean");
-                    break;
-                case TypeInteger:
-                    ptype.set("xsd:integer");
-                    break;
-                case TypeUnsignedInteger:
-                    ptype.set("xsd:integer");
-                    break;
-                case TypeReal:
-                    ptype.set("xsd:real");
-                    break;
-                case TypeSet:
-                    ptype.set("tns:EspStringArray");
-                    break;
-                case TypeDataset:
-                case TypeData:
-                    ptype.set("tns:XmlDataSet");
-                    break;
-                case TypeUnicode:
-                case TypeString:
-                    ptype.set("xsd:string");
-                    break;
-                case TypeUnknown:
-                case TypeBeginIfBlock:
-                case TypeEndIfBlock:
-                case TypeBeginRecord:
-                default:
-                    ptype.set("xsd:string");
-                    break;
-                }
-                out.appendf("<part name=\"%s\" type=\"%s\" />", varname.str(), ptype.str());
+                Owned<IConstWUResult> var = wu->getVariableByName(fields.item(i));
+                if (var)
+                    appendVariableParmInfo(parts, resultSetFactory, *var, i+1);
             }
         }
+        else
+        {
+            Owned<IConstWUResultIterator> vars = &ensureWorkUnit()->getVariables();
+            ForEach(*vars)
+                appendVariableParmInfo(parts, resultSetFactory, vars->query());
+        }
+        parts.sort(orderParts);
+        ForEachItemIn(i, parts)
+            toXML(&parts.item(i), out);
         out.append("</message>");
     }
 

esp/xslt/wsecl3_form.xsl

@@ -1169,6 +1169,9 @@ function switchInputForm()
                 </xsl:variable>
                 <xsl:variable name="inputRows">
                     <xsl:choose>
+                        <xsl:when test="$annot/@formRows">
+                            <xsl:value-of select="$annot/@formRows"/>
+                        </xsl:when>
                         <xsl:when test="$maxoccurs='unbounded'">
                             <xsl:number value="4"/>
                         </xsl:when>

roxie/ccd/ccdactivities.cpp

@@ -1022,7 +1022,7 @@ class CRoxieCsvReadActivity;
 class CRoxieXmlReadActivity;
 IInMemoryFileProcessor *createKeyedRecordProcessor(IInMemoryIndexCursor *cursor, CRoxieDiskReadActivity &owner, bool resent);
 IInMemoryFileProcessor *createUnkeyedRecordProcessor(IInMemoryIndexCursor *cursor, CRoxieDiskReadActivity &owner, bool variableDisk, IDirectReader *reader);
-IInMemoryFileProcessor *createCsvRecordProcessor(CRoxieCsvReadActivity &owner, IDirectReader *reader, bool _skipHeader, const IResolvedFile *datafile);
+IInMemoryFileProcessor *createCsvRecordProcessor(CRoxieCsvReadActivity &owner, IDirectReader *reader, bool _skipHeader, const IResolvedFile *datafile, size32_t maxRowSize);
 IInMemoryFileProcessor *createXmlRecordProcessor(CRoxieXmlReadActivity &owner, IDirectReader *reader);
 
 class CRoxieDiskReadActivity : public CRoxieDiskReadBaseActivity
@@ -1075,11 +1075,12 @@ public:
 protected:
     IHThorCsvReadArg *helper;
     const IResolvedFile *datafile;
+    size32_t maxRowSize;
 
 public:
-    CRoxieCsvReadActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory,
-                          const CSlaveActivityFactory *_aFactory, IInMemoryIndexManager *_manager, const IResolvedFile *_datafile)
-        : CRoxieDiskReadBaseActivity(_logctx, _packet, _hFactory, _aFactory, _manager, 0, 1, true), datafile(_datafile)
+    CRoxieCsvReadActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CSlaveActivityFactory *_aFactory,
+                          IInMemoryIndexManager *_manager, const IResolvedFile *_datafile, size32_t _maxRowSize)
+        : CRoxieDiskReadBaseActivity(_logctx, _packet, _hFactory, _aFactory, _manager, 0, 1, true), datafile(_datafile), maxRowSize(_maxRowSize)
     {
         onCreate();
         helper = (IHThorCsvReadArg *) basehelper;
@@ -1098,7 +1099,7 @@ public:
                     createCsvRecordProcessor(*this,
                                              manager->createReader(readPos, parallelPartNo, numParallel),
                                              packet->queryHeader().channel==1 && !resent,
-                                             varFileInfo ? varFileInfo.get() : datafile));
+                                             varFileInfo ? varFileInfo.get() : datafile, maxRowSize));
         }
         unsigned __int64 rowLimit = helper->getRowLimit();
         unsigned __int64 stopAfter = helper->getChooseNLimit();
@@ -1178,17 +1179,23 @@ public:
 
 class CRoxieCsvReadActivityFactory : public CRoxieDiskBaseActivityFactory
 {
+    size32_t maxRowSize;
+
 public:
     IMPLEMENT_IINTERFACE;
 
     CRoxieCsvReadActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
         : CRoxieDiskBaseActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
     {
+        maxRowSize = defaultMaxCsvRowSize * 1024 * 1024;
+        IConstWorkUnit *workunit = _queryFactory.queryWorkUnit();
+        if (workunit)
+            maxRowSize = workunit->getDebugValueInt(OPT_MAXCSVROWSIZE, defaultMaxCsvRowSize) * 1024 * 1024;
     }
 
     virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
     {
-        return new CRoxieCsvReadActivity(logctx, packet, helperFactory, this, manager, datafile);
+        return new CRoxieCsvReadActivity(logctx, packet, helperFactory, this, manager, datafile, maxRowSize);
     }
 
     virtual StringBuffer &toString(StringBuffer &s) const
@@ -1496,12 +1503,13 @@ protected:
     Owned<IDirectReader> reader;
     bool skipHeader;
     const IResolvedFile *datafile;
+    size32_t maxRowSize;
 
 public:
     IMPLEMENT_IINTERFACE;
 
-    CsvRecordProcessor(CRoxieCsvReadActivity &_owner, IDirectReader *_reader, bool _skipHeader, const IResolvedFile *_datafile)
-      : RecordProcessor(NULL), owner(_owner), reader(_reader), datafile(_datafile)
+    CsvRecordProcessor(CRoxieCsvReadActivity &_owner, IDirectReader *_reader, bool _skipHeader, const IResolvedFile *_datafile, size32_t _maxRowSize)
+      : RecordProcessor(NULL), owner(_owner), reader(_reader), datafile(_datafile), maxRowSize(_maxRowSize)
     {
         helper = _owner.helper;
         skipHeader = _skipHeader;
@@ -1538,7 +1546,6 @@ public:
                 break;
             }
             size32_t rowSize = 4096; // MORE - make configurable
-            size32_t maxRowSize = 10*1024*1024; // MORE - make configurable
             size32_t thisLineLength;
             loop
             {
@@ -1548,7 +1555,7 @@ public:
                 if (thisLineLength < rowSize || avail < rowSize)
                     break;
                 if (rowSize == maxRowSize)
-                    throw MakeStringException(0, "Row too big");
+                    throw MakeStringException(0, "File contained a line of length greater than %d bytes.", maxRowSize);
                 if (rowSize >= maxRowSize/2)
                     rowSize = maxRowSize;
                 else
@@ -1677,9 +1684,9 @@ protected:
     Owned<IDirectReader> reader;
 };
 
-IInMemoryFileProcessor *createCsvRecordProcessor(CRoxieCsvReadActivity &owner, IDirectReader *_reader, bool _skipHeader, const IResolvedFile *datafile)
+IInMemoryFileProcessor *createCsvRecordProcessor(CRoxieCsvReadActivity &owner, IDirectReader *_reader, bool _skipHeader, const IResolvedFile *datafile, size32_t maxRowSize)
 {
-    return new CsvRecordProcessor(owner, _reader, _skipHeader, datafile);
+    return new CsvRecordProcessor(owner, _reader, _skipHeader, datafile, maxRowSize);
 }
 
 IInMemoryFileProcessor *createXmlRecordProcessor(CRoxieXmlReadActivity &owner, IDirectReader *_reader)
@@ -4428,11 +4435,12 @@ IRoxieSlaveActivity *CRoxieFetchActivityFactory::createActivity(SlaveContextLogg
 
 class CRoxieCSVFetchActivity : public CRoxieFetchActivityBase
 {
-    CSVSplitter csvSplitter;    
+    CSVSplitter csvSplitter;
+    size32_t maxRowSize;
 
 public:
-    CRoxieCSVFetchActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CRoxieFetchActivityFactory *_aFactory, unsigned _maxColumns)
-        : CRoxieFetchActivityBase(_logctx, _packet, _hFactory, _aFactory)
+    CRoxieCSVFetchActivity(SlaveContextLogger &_logctx, IRoxieQueryPacket *_packet, HelperFactory *_hFactory, const CRoxieFetchActivityFactory *_aFactory, unsigned _maxColumns, size32_t _maxRowSize)
+        : CRoxieFetchActivityBase(_logctx, _packet, _hFactory, _aFactory), maxRowSize(_maxRowSize)
     {
         const char * quotes = NULL;
         const char * separators = NULL;
@@ -4462,7 +4470,6 @@ public:
         IHThorCsvFetchArg *h = (IHThorCsvFetchArg *) helper;
         rawStream->reset(pos);
         size32_t rowSize = 4096; // MORE - make configurable
-        size32_t maxRowSize = 10*1024*1024; // MORE - make configurable
         loop
         {
             size32_t avail;
@@ -4470,7 +4477,7 @@ public:
             if (csvSplitter.splitLine(avail, (const byte *)peek) < rowSize || avail < rowSize)
                 break;
             if (rowSize == maxRowSize)
-                throw MakeStringException(0, "Row too big");
+                throw MakeStringException(0, "File contained a line of length greater than %d bytes.", maxRowSize);
             if (rowSize >= maxRowSize/2)
                 rowSize = maxRowSize;
             else
@@ -4541,6 +4548,7 @@ void CRoxieFetchActivityBase::setPartNo(bool filechanged)
 class CRoxieCSVFetchActivityFactory : public CRoxieFetchActivityFactory
 {
     unsigned maxColumns;
+    size32_t maxRowSize;
 
 public:
     CRoxieCSVFetchActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
@@ -4550,11 +4558,15 @@ public:
         maxColumns = helper->getMaxColumns();
         ICsvParameters *csvInfo = helper->queryCsvParameters();
         assertex(!csvInfo->queryEBCDIC());
+        maxRowSize = defaultMaxCsvRowSize * 1024 * 1024;
+        IConstWorkUnit *workunit = _queryFactory.queryWorkUnit();
+        if (workunit)
+            maxRowSize = workunit->getDebugValueInt(OPT_MAXCSVROWSIZE, defaultMaxCsvRowSize) * 1024 * 1024;
     }
 
     virtual IRoxieSlaveActivity *createActivity(SlaveContextLogger &logctx, IRoxieQueryPacket *packet) const
     {
-        return new CRoxieCSVFetchActivity(logctx, packet, helperFactory, this, maxColumns);
+        return new CRoxieCSVFetchActivity(logctx, packet, helperFactory, this, maxColumns, maxRowSize);
     }
 };
 

roxie/ccd/ccdserver.cpp

@@ -19921,9 +19921,12 @@ public:
             if (helper.getFlags() & POFmaxsize)
                 outputLimit = helper.getMaxSize();
             else
-                outputLimit = workunit->getDebugValueInt("outputLimit", DALI_RESULT_LIMIT_DEFAULT);
-            if (outputLimit>DALI_RESULT_OUTPUTMAX)
-                throw MakeStringException(0, "Dali result outputs are restricted to a maximum of %d MB, the current limit is %d MB. A huge dali result usually indicates the ECL needs altering.", DALI_RESULT_OUTPUTMAX, DALI_RESULT_LIMIT_DEFAULT);
+            {
+                // In absense of OPT_OUTPUTLIMIT check pre 5.2 legacy name OPT_OUTPUTLIMIT_LEGACY
+                outputLimit = workunit->getDebugValueInt(OPT_OUTPUTLIMIT, workunit->getDebugValueInt(OPT_OUTPUTLIMIT_LEGACY, defaultDaliResultLimit));
+            }
+            if (outputLimit>defaultDaliResultOutputMax)
+                throw MakeStringException(0, "Dali result outputs are restricted to a maximum of %d MB, the current limit is %d MB. A huge dali result usually indicates the ECL needs altering.", defaultDaliResultOutputMax, defaultDaliResultLimit);
             assertex(outputLimit<=0x1000); // 32bit limit because MemoryBuffer/CMessageBuffers involved etc.
             outputLimitBytes = outputLimit * 0x100000;
         }
@@ -20799,12 +20802,13 @@ class CRoxieServerCsvReadActivity : public CRoxieServerDiskReadBaseActivity
     const char *separators;
     const char *terminators;
     const char *escapes;
+    size32_t maxRowSize;
 public:
     CRoxieServerCsvReadActivity(const IRoxieServerActivityFactory *_factory, IProbeManager *_probeManager, const RemoteActivityId &_remoteId,
                                 unsigned _numParts, bool _isLocal, bool _sorted, bool _maySkip, IInMemoryIndexManager *_manager,
-                                const char *_quotes, const char *_separators, const char *_terminators, const char *_escapes)
+                                const char *_quotes, const char *_separators, const char *_terminators, const char *_escapes, size32_t _maxRowSize)
         : CRoxieServerDiskReadBaseActivity(_factory, _probeManager, _remoteId, _numParts, _isLocal, _sorted, _maySkip, _manager),
-          quotes(_quotes), separators(_separators), terminators(_terminators), escapes(_escapes)
+          quotes(_quotes), separators(_separators), terminators(_terminators), escapes(_escapes), maxRowSize(_maxRowSize)
     {
         compoundHelper = NULL;
         readHelper = (IHThorCsvReadArg *)&helper;
@@ -20861,7 +20865,6 @@ public:
                 }
                 // MORE - there are rumours of a  csvSplitter that operates on a stream... if/when it exists, this should use it
                 size32_t rowSize = 4096; // MORE - make configurable
-                size32_t maxRowSize = 10*1024*1024; // MORE - make configurable
                 size32_t thisLineLength;
                 loop
                 {
@@ -20871,7 +20874,7 @@ public:
                     if (thisLineLength < rowSize || avail < rowSize)
                         break;
                     if (rowSize == maxRowSize)
-                        throw MakeStringException(0, "Row too big");
+                        throw MakeStringException(0, "File contained a line of length greater than %d bytes.", maxRowSize);
                     if (rowSize >= maxRowSize/2)
                         rowSize = maxRowSize;
                     else
@@ -21368,6 +21371,7 @@ public:
     const char *separators;
     const char *terminators;
     const char *escapes;
+    size32_t maxCsvRowSize;
 
     CRoxieServerDiskReadActivityFactory(unsigned _id, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory, ThorActivityKind _kind, const RemoteActivityId &_remoteId, IPropertyTree &_graphNode)
         : CRoxieServerActivityFactory(_id, _subgraphId, _queryFactory, _helperFactory, _kind), remoteId(_remoteId)
@@ -21404,6 +21408,19 @@ public:
                     manager.setown(getEmptyIndexManager());
             }
         }
+        switch (kind)
+        {
+            case TAKcsvread:
+            {
+                maxCsvRowSize = defaultMaxCsvRowSize * 1024 * 1024;
+                IConstWorkUnit *workunit = _queryFactory.queryWorkUnit();
+                if (workunit)
+                    maxCsvRowSize = workunit->getDebugValueInt(OPT_MAXCSVROWSIZE, defaultMaxCsvRowSize) * 1024 * 1024;
+                break;
+            }
+            default:
+                maxCsvRowSize = 0; // unused
+        }
     }
 
     virtual IRoxieServerActivity *createActivity(IProbeManager *_probeManager) const
@@ -21413,7 +21430,7 @@ public:
         {
         case TAKcsvread:
             return new CRoxieServerCsvReadActivity(this, _probeManager, remoteId, numParts, isLocal, sorted, maySkip, manager,
-                                                   quotes, separators, terminators, escapes);
+                                                   quotes, separators, terminators, escapes, maxCsvRowSize);
         case TAKxmlread:
         case TAKjsonread:
             return new CRoxieServerXmlReadActivity(this, _probeManager, remoteId, numParts, isLocal, sorted, maySkip, manager);

system/jlib/jqueue.tpp

@@ -236,9 +236,9 @@ public:
         }
         return (unsigned)-1;
     }
-    void dequeue(BASE *e)
+    BASE *dequeue(BASE *e)
     {
-        dequeue(find(e));
+        return dequeue(find(e));
     }
     inline unsigned ordinality() const { return num; }
 };

testing/regress/ecl/formatstored.ecl

@@ -0,0 +1,58 @@
+/*##############################################################################
+
+    HPCC SYSTEMS software Copyright (C) 2014 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.
+############################################################################## */
+
+//sequence should place these storeds in sensible order on the form, in xsd and wsdl
+//  field sizes should gradually increase fore each type on form
+
+unsigned2 u2 := 0 : stored('u2', few, format(fieldwidth(2), sequence(12)));
+integer2 i2 := 0 : stored('i2', format(fieldwidth(2), sequence(2)));
+unsigned7 u7 := 0 : stored('u7', format(fieldwidth(7), sequence(17)));
+integer3 i3 := 0 : stored('i3', format(fieldwidth(3), sequence(3)));
+integer1 i1 := 0 : stored('i1', format(fieldwidth(1), sequence(1)));
+unsigned6 u6 := 0 : stored('u6', format(fieldwidth(6), sequence(16)));
+integer6 i6 := 0 : stored('i6', format(fieldwidth(6), sequence(6)));
+unsigned1 u1 := 0 : stored('u1', format(fieldwidth(1), sequence(11)));
+integer7 i7 := 0 : stored('i7', format(fieldwidth(7), sequence(7)));
+integer8 i8 := 0 : stored('i8', format(fieldwidth(8), sequence(8)));
+string s1 := 'how now brown cow' : stored('s1', format(fieldwidth(40), fieldheight(10), sequence(20)));
+unsigned4 u4 := 0 : stored('u4', format(fieldwidth(4), sequence(14)));
+integer5 i5 := 0 : stored('i5', format(fieldwidth(5), sequence(5)));
+unsigned5 u5 := 0 : stored('u5', format(fieldwidth(5), sequence(15)));
+integer4 i4 := 0 : stored('i4', format(fieldwidth(4), sequence(4)));
+unsigned3 u3 := 0 : stored('u3', format(fieldwidth(3), sequence(13)));
+unsigned8 u8 := 0 : stored('u8', format(fieldwidth(8), sequence(18)));
+
+output (i1, named('i1'));
+output (i2, named('i2'));
+output (i3, named('i3'));
+output (i4, named('i4'));
+output (i5, named('i5'));
+output (i6, named('i6'));
+output (i7, named('i7'));
+output (i8, named('i8'));
+
+output (u1, named('u1'));
+output (u2, named('u2'));
+output (u3, named('u3'));
+output (u4, named('u4'));
+output (u5, named('u5'));
+output (u6, named('u6'));
+output (u7, named('u7'));
+output (u8, named('u8'));
+
+output (s1, named('s1'));
+

testing/regress/ecl/key/formatstored.xml

@@ -0,0 +1,51 @@
+<Dataset name='i1'>
+ <Row><i1>0</i1></Row>
+</Dataset>
+<Dataset name='i2'>
+ <Row><i2>0</i2></Row>
+</Dataset>
+<Dataset name='i3'>
+ <Row><i3>0</i3></Row>
+</Dataset>
+<Dataset name='i4'>
+ <Row><i4>0</i4></Row>
+</Dataset>
+<Dataset name='i5'>
+ <Row><i5>0</i5></Row>
+</Dataset>
+<Dataset name='i6'>
+ <Row><i6>0</i6></Row>
+</Dataset>
+<Dataset name='i7'>
+ <Row><i7>0</i7></Row>
+</Dataset>
+<Dataset name='i8'>
+ <Row><i8>0</i8></Row>
+</Dataset>
+<Dataset name='u1'>
+ <Row><u1>0</u1></Row>
+</Dataset>
+<Dataset name='u2'>
+ <Row><u2>0</u2></Row>
+</Dataset>
+<Dataset name='u3'>
+ <Row><u3>0</u3></Row>
+</Dataset>
+<Dataset name='u4'>
+ <Row><u4>0</u4></Row>
+</Dataset>
+<Dataset name='u5'>
+ <Row><u5>0</u5></Row>
+</Dataset>
+<Dataset name='u6'>
+ <Row><u6>0</u6></Row>
+</Dataset>
+<Dataset name='u7'>
+ <Row><u7>0</u7></Row>
+</Dataset>
+<Dataset name='u8'>
+ <Row><u8>0</u8></Row>
+</Dataset>
+<Dataset name='s1'>
+ <Row><s1>how now brown cow</s1></Row>
+</Dataset>

testing/regress/ecl/key/memcachedtest.xml

@@ -1,5 +1,3 @@
- <Info><Code>10001</Code><Source></Source><Message>Memcached Plugin: The key &apos;pi&apos; to fetch is of type REAL, not INTEGER as requested.
-</Message></Info>
 <Dataset name='Result 1'>
  <Row><Result_1>true</Result_1></Row>
 </Dataset>
@@ -68,4 +66,4 @@
 </Dataset>
 <Dataset name='Result 23'>
  <Row><Result_23>Nonexistent</Result_23></Row>
-</Dataset>
+</Dataset>

testing/regress/ecl/key/webservice.xml

@@ -0,0 +1,51 @@
+<Dataset name='i1'>
+ <Row><i1>0</i1></Row>
+</Dataset>
+<Dataset name='i2'>
+ <Row><i2>0</i2></Row>
+</Dataset>
+<Dataset name='i3'>
+ <Row><i3>0</i3></Row>
+</Dataset>
+<Dataset name='i4'>
+ <Row><i4>0</i4></Row>
+</Dataset>
+<Dataset name='i5'>
+ <Row><i5>0</i5></Row>
+</Dataset>
+<Dataset name='i6'>
+ <Row><i6>0</i6></Row>
+</Dataset>
+<Dataset name='i7'>
+ <Row><i7>0</i7></Row>
+</Dataset>
+<Dataset name='i8'>
+ <Row><i8>0</i8></Row>
+</Dataset>
+<Dataset name='u1'>
+ <Row><u1>0</u1></Row>
+</Dataset>
+<Dataset name='u2'>
+ <Row><u2>0</u2></Row>
+</Dataset>
+<Dataset name='u3'>
+ <Row><u3>0</u3></Row>
+</Dataset>
+<Dataset name='u4'>
+ <Row><u4>0</u4></Row>
+</Dataset>
+<Dataset name='u5'>
+ <Row><u5>0</u5></Row>
+</Dataset>
+<Dataset name='u6'>
+ <Row><u6>0</u6></Row>
+</Dataset>
+<Dataset name='u7'>
+ <Row><u7>0</u7></Row>
+</Dataset>
+<Dataset name='u8'>
+ <Row><u8>0</u8></Row>
+</Dataset>
+<Dataset name='s1'>
+ <Row><s1>how now brown cow</s1></Row>
+</Dataset>

testing/regress/ecl/webservice.ecl

@@ -0,0 +1,60 @@
+/*##############################################################################
+
+    HPCC SYSTEMS software Copyright (C) 2014 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.
+############################################################################## */
+
+string descrText := 'only the listed fields should show in form, xsd, and wsdl<br/>and in the given order<br/><br/>';
+string helpText := 'Enter some values and hit submit';
+
+#webservice(fields('u1', 'i1', 'u2', 'i2', 's1'), help(helpText), description(descrText));
+
+unsigned2 u2 := 0 : stored('u2', few, format(fieldwidth(2), sequence(12)));
+integer2 i2 := 0 : stored('i2', format(fieldwidth(2), sequence(2)));
+unsigned7 u7 := 0 : stored('u7', format(fieldwidth(7), sequence(17)));
+integer3 i3 := 0 : stored('i3', format(fieldwidth(3), sequence(3)));
+integer1 i1 := 0 : stored('i1', format(fieldwidth(1), sequence(1)));
+unsigned6 u6 := 0 : stored('u6', format(fieldwidth(6), sequence(16)));
+integer6 i6 := 0 : stored('i6', format(fieldwidth(6), sequence(6)));
+unsigned1 u1 := 0 : stored('u1', format(fieldwidth(1), sequence(11)));
+integer7 i7 := 0 : stored('i7', format(fieldwidth(7), sequence(7)));
+integer8 i8 := 0 : stored('i8', format(fieldwidth(8), sequence(8)));
+string s1 := 'how now brown cow' : stored('s1', format(fieldwidth(40), fieldheight(10), sequence(20)));
+unsigned4 u4 := 0 : stored('u4', format(fieldwidth(4), sequence(14)));
+integer5 i5 := 0 : stored('i5', format(fieldwidth(5), sequence(5)));
+unsigned5 u5 := 0 : stored('u5', format(fieldwidth(5), sequence(15)));
+integer4 i4 := 0 : stored('i4', format(fieldwidth(4), sequence(4)));
+unsigned3 u3 := 0 : stored('u3', format(fieldwidth(3), sequence(13)));
+unsigned8 u8 := 0 : stored('u8', format(fieldwidth(8), sequence(18)));
+
+output (i1, named('i1'));
+output (i2, named('i2'));
+output (i3, named('i3'));
+output (i4, named('i4'));
+output (i5, named('i5'));
+output (i6, named('i6'));
+output (i7, named('i7'));
+output (i8, named('i8'));
+
+output (u1, named('u1'));
+output (u2, named('u2'));
+output (u3, named('u3'));
+output (u4, named('u4'));
+output (u5, named('u5'));
+output (u6, named('u6'));
+output (u7, named('u7'));
+output (u8, named('u8'));
+
+output (s1, named('s1'));
+

thorlcr/activities/csvread/thcsvrslave.cpp

@@ -55,13 +55,13 @@ class CCsvReadSlaveActivity : public CDiskReadSlaveActivityBase, public CThorDat
         CRC32 inputCRC;
         bool readFinished;
         offset_t localOffset;
+        size32_t maxRowSize;
 
         unsigned splitLine()
         {
             if (inputStream->eos())
                 return 0;
             size32_t minRequired = 4096; // MORE - make configurable
-            size32_t maxRowSize = 10*1024*1024; // MORE - make configurable
             size32_t thisLineLength;
             loop
             {
@@ -86,6 +86,7 @@ class CCsvReadSlaveActivity : public CDiskReadSlaveActivityBase, public CThorDat
             //Initialise information...
             ICsvParameters * csvInfo = activity.helper->queryCsvParameters();
             csvSplitter.init(activity.helper->getMaxColumns(), csvInfo, activity.csvQuote, activity.csvSeparate, activity.csvTerminate, activity.csvEscape);
+            maxRowSize = activity.getOptInt(OPT_MAXCSVROWSIZE, defaultMaxCsvRowSize) * 1024 * 1024;
         }
         virtual void setPart(IPartDescriptor *partDesc, unsigned partNoSerialized)
         {

thorlcr/activities/nsplitter/thnsplitterslave.cpp

@@ -61,12 +61,13 @@ public:
 // NSplitterSlaveActivity
 //
 
-class NSplitterSlaveActivity : public CSlaveActivity
+class NSplitterSlaveActivity : public CSlaveActivity, implements ISharedSmartBufferCallback
 {
     bool spill;
     bool eofHit;
     bool inputsConfigured;
-    CriticalSection startLock;
+    bool writeBlocked, pagedOut;
+    CriticalSection startLock, writeAheadCrit, writeBlockedCrit;
     unsigned nstopped;
     rowcount_t recsReady;
     IThorDataLink *input;
@@ -74,44 +75,29 @@ class NSplitterSlaveActivity : public CSlaveActivity
     Owned<IException> startException, writeAheadException;
     Owned<ISharedSmartBuffer> smartBuf;
 
+    // NB: CWriter only used by 'balanced' splitter, which blocks write when too far ahead
     class CWriter : public CSimpleInterface, IThreaded
     {
         NSplitterSlaveActivity &parent;
         CThreadedPersistent threaded;
-        Semaphore sem;
         bool stopped;
-        rowcount_t writerMax;
-        unsigned requestN;
-        SpinLock recLock;
+        rowcount_t current;
 
     public:
         CWriter(NSplitterSlaveActivity &_parent) : parent(_parent), threaded("CWriter", this)
         {
+            current = 0;
             stopped = true;
-            writerMax = 0;
-            requestN = 1000;
         }
         ~CWriter() { stop(); }
         virtual void main()
         {
-            loop
-            {
-                sem.wait();
-                if (stopped) break;
-                rowcount_t request;
-                {
-                    SpinBlock block(recLock);
-                    request = writerMax;
-                }
-                parent.writeahead(request, stopped);
-                if (parent.eofHit)
-                    break;
-            }
+            while (!stopped && !parent.eofHit)
+                current = parent.writeahead(current, stopped);
         }
         void start()
         {
             stopped = false;
-            writerMax = 0;
             threaded.start();
         }
         virtual void stop()
@@ -119,21 +105,9 @@ class NSplitterSlaveActivity : public CSlaveActivity
             if (!stopped)
             {
                 stopped = true;
-                sem.signal();
                 threaded.join();
             }
         }
-        void more(rowcount_t &max) // called if output hit it's max, returns new max
-        {
-            if (stopped) return;
-            SpinBlock block(recLock);
-            if (writerMax <= max)
-            {
-                writerMax += requestN;
-                sem.signal();
-            }
-            max = writerMax;
-        }
     } writer;
     class CNullInput : public CSplitterOutputBase
     {
@@ -232,13 +206,12 @@ class NSplitterSlaveActivity : public CSlaveActivity
 public:
     IMPLEMENT_IINTERFACE_USING(CSimpleInterface);
 
-    NSplitterSlaveActivity(CGraphElementBase *container) 
-        : CSlaveActivity(container), writer(*this)
+    NSplitterSlaveActivity(CGraphElementBase *container) : CSlaveActivity(container), writer(*this)
     {
         spill = false;
         input = NULL;
         nstopped = 0;
-        eofHit = inputsConfigured = false;
+        eofHit = inputsConfigured = writeBlocked = pagedOut = false;
         recsReady = 0;
     }
     void ensureInputsConfigured()
@@ -287,6 +260,7 @@ public:
         grouped = false;
         eofHit = false;
         recsReady = 0;
+        writeBlocked = false;
         if (inputsConfigured)
         {
             // ensure old inputs cleared, to avoid being reused before re-setup on subsequent executions
@@ -309,18 +283,14 @@ public:
         else
             spill = dV>0;
     }
-    inline void more(rowcount_t &max)
-    {
-        ActivityTimer t(totalCycles, queryTimeActivities());
-        writer.more(max);
-    }
     void prepareInput(unsigned output)
     {
         CriticalBlock block(startLock);
         if (!input)
         {
             input = inputs.item(0);
-            try {
+            try
+            {
                 startInput(input);
                 grouped = input->isGrouped();
                 nstopped = container.connectedOutputs.getCount();
@@ -348,9 +318,11 @@ public:
                             smartBuf->queryOutput(o)->stop();
                     }
                 }
-                writer.start();
+                if (!spill)
+                    writer.start(); // writer keeps writing ahead as much as possible, the readahead impl. will block when has too much
             }
-            catch (IException *e) { 
+            catch (IException *e)
+            {
                 startException.setown(e); 
             }
         }
@@ -362,15 +334,37 @@ public:
             throw LINK(writeAheadException);
         return row.getClear();
     }
-    void writeahead(const rowcount_t &requested, const bool &stopped)
+    rowcount_t writeahead(rowcount_t current, const bool &stopped)
     {
-        if (eofHit)
-            return;
-        
+        // NB: readers call writeahead, which will block others
+        CriticalBlock b(writeAheadCrit);
+        loop
+        {
+            if (eofHit)
+                return recsReady;
+            if (current < recsReady)
+                return recsReady;
+            else if (writeBlocked)
+            {
+                // NB: When here, writeAheadCrit has been released _and_ writeBlockedCrit will be held until the writer is unblocked
+                {
+                    CriticalUnblock ub(writeAheadCrit);
+                    {
+                        CriticalBlock b(writeBlockedCrit); // This will block until the thread that is writing ahead has done
+                        // Once writeBlockCrit gained, writeBlocked is always 'false'
+                    }
+                }
+                // recsReady or eofHit will have been updated by the blocking thread by now, loop and re-check
+            }
+            else
+                break;
+        }
+        ActivityTimer t(totalCycles, queryTimeActivities());
+        pagedOut = false;
         OwnedConstThorRow row;
         loop
         {
-            if (abortSoon || stopped || requested<recsReady)
+            if (abortSoon || stopped || pagedOut)
                 break;
             try
             {
@@ -380,8 +374,8 @@ public:
                     row.setown(input->nextRow());
                     if (row)
                     {
+                        smartBuf->putRow(NULL, this);
                         ++recsReady;
-                        smartBuf->putRow(NULL);
                     }
                 }
             }
@@ -393,9 +387,10 @@ public:
                 smartBuf->flush(); // signals no more rows will be written.
                 break;
             }
+            smartBuf->putRow(row.getClear(), this); // can block if mem limited, but other readers can progress which is the point
             ++recsReady;
-            smartBuf->putRow(row.getClear()); // can block if mem limited, but other readers can progress which is the point
         }
+        return recsReady;
     }
     void inputStopped()
     {
@@ -422,7 +417,20 @@ public:
         }
         return _totalCycles;
     }
-
+// ISharedSmartBufferCallback impl.
+    virtual void paged() { pagedOut = true; }
+    virtual void blocked()
+    {
+        writeBlockedCrit.enter(); // enter before unblocking writeahead
+        writeBlocked = true; // Prevent other users getting beyond checking recsReady in writeahead()
+        writeAheadCrit.leave();
+    }
+    virtual void unblocked()
+    {
+        writeAheadCrit.enter();
+        writeBlocked = false;
+        writeBlockedCrit.leave(); // leave after re-blocking writer
+    }
 friend class CInputWrapper;
 friend class CSplitterOutput;
 friend class CWriter;
@@ -457,7 +465,7 @@ void CSplitterOutput::stop()
 const void *CSplitterOutput::nextRow()
 {
     if (rec == max)
-        activity.more(max);
+        max = activity.writeahead(max, activity.queryAbortSoon());
     ActivityTimer t(totalCycles, activity.queryTimeActivities());
     const void *row = activity.nextRow(output); // pass ptr to max if need more
     ++rec;

thorlcr/activities/wuidwrite/thwuidwrite.cpp

@@ -53,9 +53,10 @@ public:
     virtual void init()
     {
         CMasterActivity::init();
-        workunitWriteLimit = activityMaxSize ? activityMaxSize : getOptInt(THOROPT_OUTPUTLIMIT, DEFAULT_WUIDWRITE_LIMIT);
-        if (workunitWriteLimit>DALI_RESULT_OUTPUTMAX)
-            throw MakeActivityException(this, 0, "Configured max result size, %d MB, exceeds absolute max limit of %d MB. A huge Dali result usually indicates the ECL needs altering.", workunitWriteLimit, DALI_RESULT_OUTPUTMAX);
+        // In absense of OPT_OUTPUTLIMIT check pre 5.2 legacy name OPT_OUTPUTLIMIT_LEGACY
+        workunitWriteLimit = activityMaxSize ? activityMaxSize : getOptInt(OPT_OUTPUTLIMIT, getOptInt(OPT_OUTPUTLIMIT_LEGACY, defaultDaliResultLimit));
+        if (workunitWriteLimit>defaultDaliResultOutputMax)
+            throw MakeActivityException(this, 0, "Configured max result size, %d MB, exceeds absolute max limit of %d MB. A huge Dali result usually indicates the ECL needs altering.", workunitWriteLimit, defaultDaliResultOutputMax);
         assertex(workunitWriteLimit<=0x1000); // 32bit limit because MemoryBuffer/CMessageBuffers involved etc.
         workunitWriteLimit *= 0x100000;
     }

thorlcr/master/thmastermain.cpp

@@ -124,7 +124,7 @@ class CRegistryServer : public CSimpleInterface
                 }
                 RegistryCode code;
                 msg.read((int &)code);
-                if (!rc_deregister == code)
+                if (rc_deregister != code)
                     throwUnexpected();
                 registry.deregisterNode(sender);
             }

thorlcr/thorutil/thbuf.cpp

@@ -670,14 +670,23 @@ IRowWriterMultiReader *createOverflowableBuffer(CActivityBase &activity, IRowInt
 #define VALIDATELT(LHS, RHS) if ((LHS)>=(RHS)) { StringBuffer s("FAIL(LT) - LHS="); s.append(LHS).append(", RHS=").append(RHS); PROGLOG("%s", s.str()); throwUnexpected(); }
 
 //#define TRACE_WRITEAHEAD
-class CRowSet : public CSimpleInterface
+class CSharedWriteAheadBase;
+class CRowSet : public CSimpleInterface, implements IInterface
 {
     unsigned chunk;
     CThorExpandingRowArray rows;
+    CSharedWriteAheadBase &sharedWriteAhead;
+    mutable SpinLock lock;
+    mutable CriticalSection crit;
 public:
-    CRowSet(CActivityBase &activity, unsigned _chunk) : rows(activity, &activity, true), chunk(_chunk)
+    CRowSet(CSharedWriteAheadBase &_sharedWriteAhead, unsigned _chunk, unsigned maxRows);
+    virtual void Link() const
     {
+        CSimpleInterface::Link();
     }
+    virtual bool Release() const;
+    void clear() { rows.clearRows(); }
+    void setChunk(unsigned _chunk) { chunk = _chunk; } 
     void reset(unsigned _chunk)
     {
         chunk = _chunk;
@@ -697,6 +706,8 @@ class Chunk : public CInterface
 public:
     offset_t offset;
     size32_t size;
+    Linked<CRowSet> rowSet;
+    Chunk(CRowSet *_rowSet) : rowSet(_rowSet), offset(0), size(0) { }
     Chunk(offset_t _offset, size_t _size) : offset(_offset), size(_size) { }
     Chunk(const Chunk &other) : offset(other.offset), size(other.size) { }
     bool operator==(Chunk const &other) const { return size==other.size && offset==other.offset; }
@@ -724,6 +735,7 @@ int chunkSizeCompare2(Chunk *lhs, Chunk *rhs)
     return (int)lhs->size - (int)rhs->size;
 }
 
+#define MIN_POOL_CHUNKS 10
 class CSharedWriteAheadBase : public CSimpleInterface, implements ISharedSmartBuffer
 {
     size32_t totalOutChunkSize;
@@ -731,7 +743,16 @@ class CSharedWriteAheadBase : public CSimpleInterface, implements ISharedSmartBu
     rowcount_t rowsWritten;
     IArrayOf<IRowStream> outputs;
     unsigned readersWaiting;
+    mutable IArrayOf<CRowSet> cachedRowSets;
+    CriticalSection rowSetCacheCrit;
 
+    void reuse(CRowSet *rowset)
+    {
+        rowset->clear();
+        CriticalBlock b(rowSetCacheCrit);
+        if (cachedRowSets.ordinality() < (outputs.ordinality()*2))
+            cachedRowSets.append(*LINK(rowset));
+    }
     virtual void init()
     {
         stopped = false;
@@ -763,11 +784,6 @@ class CSharedWriteAheadBase : public CSimpleInterface, implements ISharedSmartBu
                 queryCOutput(o).wakeRead(); // if necessary
         }
     }
-    inline const rowcount_t &readerWait(const rowcount_t &rowsRead)
-    {
-        ++readersWaiting;
-        return rowsWritten;
-    }
     CRowSet *loadMore(unsigned output)
     {
         // needs to be called in crit
@@ -819,32 +835,6 @@ protected:
             eof = true;
             parent.stopOutput(output);
         }
-        inline const rowcount_t readerWait(const rowcount_t &rowsRead)
-        {
-            if (rowsRead == parent.rowsWritten)
-            {
-                if (parent.stopped || parent.writeAtEof)
-                    return 0;
-                readerWaiting = true;
-#ifdef TRACE_WRITEAHEAD
-                ActPrintLogEx(&parent.activity->queryContainer(), thorlog_all, MCdebugProgress, "readerWait(%d)", output);
-#endif
-                const rowcount_t &rowsWritten = parent.readerWait(rowsRead);
-                {
-                    CriticalUnblock b(parent.crit);
-                    unsigned mins=0;
-                    loop
-                    {
-                        if (readWaitSem.wait(60000))
-                            break; // NB: will also be signal if aborting
-                        ActPrintLog(parent.activity, "output %d @ row # %"RCPF"d, has been blocked for %d minute(s)", output, rowsRead, ++mins);
-                    }
-                }
-                if (parent.isEof(rowsRead))
-                    return 0;
-            }
-            return parent.rowsWritten;
-        }
         inline void wakeRead()
         {
             if (readerWaiting)
@@ -868,6 +858,7 @@ protected:
             init();
             outputOwnedRows.clear();
         }
+        inline unsigned queryOutput() const { return output; }
         inline CRowSet *queryRowSet() { return rowSet; }
         const void *nextRow()
         {
@@ -878,13 +869,13 @@ protected:
                 CriticalBlock b(parent.crit);
                 if (!rowSet || (row == (rowsInRowSet = rowSet->getRowCount())))
                 {
-                    rowcount_t totalRows = readerWait(rowsRead);
+                    rowcount_t totalRows = parent.readerWait(*this, rowsRead);
                     if (0 == totalRows)
                     {
                         doStop();
                         return NULL;
                     }
-                    if (!rowSet || (row == (rowsInRowSet = rowSet->getRowCount()))) // maybe have caught up in same rowSet
+                    if (!rowSet || (row == (rowsInRowSet = rowSet->getRowCount()))) // may have caught up in same rowSet
                     {
                         Owned<CRowSet> newRows = parent.loadMore(output);
                         if (rowSet)
@@ -897,7 +888,6 @@ protected:
                 }
             }
             rowsRead++;
-            SpinBlock b(parent.spin);
             const void *retrow = rowSet->getRow(row++);
             return retrow;
         }
@@ -911,12 +901,53 @@ protected:
     CActivityBase *activity;
     size32_t minChunkSize;
     unsigned lowestChunk, lowestOutput, outputCount, totalChunksOut;
+    rowidx_t maxRows;
     bool stopped;
     Owned<CRowSet> inMemRows;
     CriticalSection crit;
-    SpinLock spin;
     Linked<IOutputMetaData> meta;
+    QueueOf<CRowSet, false> chunkPool;
+    unsigned maxPoolChunks;
 
+    inline const rowcount_t readerWait(COutput &output, const rowcount_t rowsRead)
+    {
+        if (rowsRead == rowsWritten)
+        {
+            if (stopped || writeAtEof)
+                return 0;
+            output.readerWaiting = true;
+#ifdef TRACE_WRITEAHEAD
+            ActPrintLogEx(&activity->queryContainer(), thorlog_all, MCdebugProgress, "readerWait(%d)", output.queryOutput());
+#endif
+            ++readersWaiting;
+            {
+                CriticalUnblock b(crit);
+                unsigned mins=0;
+                loop
+                {
+                    if (output.readWaitSem.wait(60000))
+                        break; // NB: will also be signal if aborting
+                    ActPrintLog(activity, "output %d @ row # %"RCPF"d, has been blocked for %d minute(s)", output.queryOutput(), rowsRead, ++mins);
+                }
+            }
+            if (isEof(rowsRead))
+                return 0;
+        }
+        return rowsWritten;
+    }
+    CRowSet *newRowSet(unsigned chunk)
+    {
+        {
+            CriticalBlock b(rowSetCacheCrit);
+            if (cachedRowSets.ordinality())
+            {
+                CRowSet *rowSet = &cachedRowSets.popGet();
+                rowSet->setChunk(chunk);
+                return rowSet;
+            }
+        }
+        return new CRowSet(*this, chunk, maxRows);
+    }
     inline COutput &queryCOutput(unsigned i) { return (COutput &) outputs.item(i); }
     inline unsigned getLowest()
     {
@@ -1001,7 +1032,7 @@ protected:
     }
     virtual void freeOffsetChunk(unsigned chunk) = 0;
     virtual CRowSet *readRows(unsigned output, unsigned chunk) = 0;
-    virtual void flushRows() = 0;
+    virtual void flushRows(ISharedSmartBufferCallback *callback) = 0;
     virtual size32_t rowSize(const void *row) = 0;
 public:
 
@@ -1018,13 +1049,15 @@ public:
             if (minChunkSize > 0x10000)
                 minChunkSize += 2*(minSize+1);
         }
+        maxRows = (minChunkSize / minSize) + 1;
         outputCount = _outputCount;
         unsigned c=0;
         for (; c<outputCount; c++)
         {
             outputs.append(* new COutput(*this, c));
         }
-        inMemRows.setown(new CRowSet(*activity, 0));
+        inMemRows.setown(newRowSet(0));
+        maxPoolChunks = MIN_POOL_CHUNKS;
     }
     ~CSharedWriteAheadBase()
     {
@@ -1052,7 +1085,7 @@ public:
     }
 
 // ISharedSmartBuffer
-    virtual void putRow(const void *row)
+    virtual void putRow(const void *row, ISharedSmartBufferCallback *callback)
     {
         if (stopped)
         {
@@ -1061,27 +1094,38 @@ public:
         }
         unsigned len=rowSize(row);
         CriticalBlock b(crit);
+        bool paged = false;
         if (totalOutChunkSize >= minChunkSize) // chunks required to be at least minChunkSize
         {
             unsigned reader=anyReaderBehind();
             if (NotFound != reader)
-                flushRows();
-            inMemRows.setown(new CRowSet(*activity, ++totalChunksOut));
+                flushRows(callback);
+            inMemRows.setown(newRowSet(++totalChunksOut));
 #ifdef TRACE_WRITEAHEAD
             totalOutChunkSize = sizeof(unsigned);
 #else
             totalOutChunkSize = 0;
 #endif
+            /* If callback used to signal paged out, only signal readers on page event,
+             * This is to minimize time spent by fast readers constantly catching up and waiting and getting woken up per record
+             */
+            if (callback)
+            {
+                paged = true;
+                callback->paged();
+            }
         }
-        {
-            SpinBlock b(spin);
-            inMemRows->addRow(row);
-        }
+        inMemRows->addRow(row);
 
         totalOutChunkSize += len;
         rowsWritten++; // including NULLs(eogs)
 
-        signalReaders();
+        if (!callback || paged)
+            signalReaders();
+    }
+    virtual void putRow(const void *row)
+    {
+        return putRow(row, NULL);
     }
     virtual void flush()
     {
@@ -1113,8 +1157,25 @@ public:
         inMemRows->reset(0);
     }
 friend class COutput;
+friend class CRowSet;
 };
 
+CRowSet::CRowSet(CSharedWriteAheadBase &_sharedWriteAhead, unsigned _chunk, unsigned maxRows)
+    : sharedWriteAhead(_sharedWriteAhead), rows(*_sharedWriteAhead.activity, _sharedWriteAhead.activity, true, stableSort_none, true, maxRows), chunk(_chunk)
+{
+}
+
+bool CRowSet::Release() const
+{
+    {
+        // NB: Occasionally, >1 thread may be releasing a CRowSet concurrently and miss a opportunity to reuse, but that's ok.
+        SpinBlock b(lock);
+        if (!IsShared())
+            sharedWriteAhead.reuse((CRowSet *)this);
+    }
+    return CSimpleInterface::Release();
+}
+
 class CSharedWriteAheadDisk : public CSharedWriteAheadBase
 {
     IDiskUsage *iDiskUsage;
@@ -1226,7 +1287,7 @@ class CSharedWriteAheadDisk : public CSharedWriteAheadBase
                     return chunk.getClear();
                 }
 #ifdef TRACE_WRITEAHEAD
-                ActPrintLogEx(&activity->queryContainer(), thorlog_all, MCdebugProgress, "getOutOffset: got [free] offset = %"I64F"d, size=%d", diskChunk.offset, diskChunk.size);
+                ActPrintLogEx(&activity->queryContainer(), thorlog_all, MCdebugProgress, "getOutOffset: got [free] offset = %"I64F"d, size=%d", nextChunk->offset, nextChunk->size);
 #endif
                 freeChunksSized.remove(nextPos);
                 freeChunks.zap(*nextChunk);
@@ -1280,6 +1341,15 @@ class CSharedWriteAheadDisk : public CSharedWriteAheadBase
         // chunk(unused here) is nominal sequential page #, savedChunks is page # in diskfile
         assertex(savedChunks.ordinality());
         Owned<Chunk> freeChunk = savedChunks.dequeue();
+        if (freeChunk->rowSet)
+        {
+            Owned<CRowSet> rowSet = chunkPool.dequeue(freeChunk->rowSet);
+#ifdef TRACE_WRITEAHEAD
+            ActPrintLogEx(&activity->queryContainer(), thorlog_all, MCdebugProgress, "freeOffsetChunk (chunk=%d) chunkPool, savedChunks size=%d, chunkPool size=%d", rowSet->queryChunk(), savedChunks.ordinality(), chunkPool.ordinality());
+#endif
+            VALIDATEEQ(chunk, rowSet->queryChunk());
+            return;
+        }
         unsigned nmemb = freeChunks.ordinality();
         if (0 == nmemb)
             addFreeChunk(freeChunk);
@@ -1329,63 +1399,88 @@ class CSharedWriteAheadDisk : public CSharedWriteAheadBase
             }
         }
         Chunk &chunk = *savedChunks.item(whichChunk);
-        Owned<ISerialStream> stream = createFileSerialStream(spillFileIO, chunk.offset);
+        Owned<CRowSet> rowSet;
+        if (chunk.rowSet)
+        {
+            rowSet.set(chunk.rowSet);
 #ifdef TRACE_WRITEAHEAD
-        unsigned diskChunkNum;
-        stream->get(sizeof(diskChunkNum), &diskChunkNum);
-        VALIDATEEQ(diskChunkNum, currentChunkNum);
+            ActPrintLogEx(&activity->queryContainer(), thorlog_all, MCdebugProgress, "readRows (chunk=%d) output: %d, savedChunks size=%d, chunkPool size=%d, currentChunkNum=%d, whichChunk=%d", rowSet->queryChunk(), output, savedChunks.ordinality(), chunkPool.ordinality(), currentChunkNum, whichChunk);
 #endif
-        CThorStreamDeserializerSource ds(stream);
-        Owned<CRowSet> rowSet = new CRowSet(*activity, currentChunkNum);
-        loop
-        {   
-            byte b;
-            ds.read(sizeof(b),&b);
-            if (!b)
-                break;
-            if (1==b)
+            VALIDATEEQ(rowSet->queryChunk(), currentChunkNum);
+        }
+        else
+        {
+            Owned<ISerialStream> stream = createFileSerialStream(spillFileIO, chunk.offset);
+#ifdef TRACE_WRITEAHEAD
+            unsigned diskChunkNum;
+            stream->get(sizeof(diskChunkNum), &diskChunkNum);
+            VALIDATEEQ(diskChunkNum, currentChunkNum);
+#endif
+            CThorStreamDeserializerSource ds(stream);
+            rowSet.setown(newRowSet(currentChunkNum));
+            loop
             {
-                RtlDynamicRowBuilder rowBuilder(allocator);
-                size32_t sz = deserializer->deserialize(rowBuilder, ds);
-                rowSet->addRow(rowBuilder.finalizeRowClear(sz));
+                byte b;
+                ds.read(sizeof(b),&b);
+                if (!b)
+                    break;
+                if (1==b)
+                {
+                    RtlDynamicRowBuilder rowBuilder(allocator);
+                    size32_t sz = deserializer->deserialize(rowBuilder, ds);
+                    rowSet->addRow(rowBuilder.finalizeRowClear(sz));
+                }
+                else if (2==b)
+                    rowSet->addRow(NULL);
             }
-            else if (2==b)
-                rowSet->addRow(NULL);
         }
         return rowSet.getClear();
     }
-    virtual void flushRows()
+    virtual void flushRows(ISharedSmartBufferCallback *callback)
     {
         // NB: called in crit
-        MemoryBuffer mb;
-        mb.ensureCapacity(minChunkSize); // starting size/could be more if variable and bigger
+        Owned<Chunk> chunk;
+        if (chunkPool.ordinality() < maxPoolChunks)
+        {
+            chunk.setown(new Chunk(inMemRows));
+            chunkPool.enqueue(inMemRows.getLink());
 #ifdef TRACE_WRITEAHEAD
-        mb.append(inMemRows->queryChunk()); // for debug purposes only
+            ActPrintLogEx(&activity->queryContainer(), thorlog_all, MCdebugProgress, "flushRows (chunk=%d) savedChunks size=%d, chunkPool size=%d", inMemRows->queryChunk(), savedChunks.ordinality()+1, chunkPool.ordinality());
 #endif
-        CMemoryRowSerializer mbs(mb);
-        unsigned r=0;
-        for (;r<inMemRows->getRowCount();r++)
+        }
+        else
         {
-            OwnedConstThorRow row = inMemRows->getRow(r);
-            if (row)
+            /* It might be worth adding a heuristic here, to estimate time for readers to catch up, vs time spend writing and reading.
+             * Could block for readers to catch up if cost of writing/reads outweighs avg cost of catch up...
+             */
+            MemoryBuffer mb;
+            mb.ensureCapacity(minChunkSize); // starting size/could be more if variable and bigger
+#ifdef TRACE_WRITEAHEAD
+            mb.append(inMemRows->queryChunk()); // for debug purposes only
+#endif
+            CMemoryRowSerializer mbs(mb);
+            unsigned r=0;
+            for (;r<inMemRows->getRowCount();r++)
             {
-                mb.append((byte)1);
-                serializer->serialize(mbs,(const byte *)row.get());
+                OwnedConstThorRow row = inMemRows->getRow(r);
+                if (row)
+                {
+                    mb.append((byte)1);
+                    serializer->serialize(mbs,(const byte *)row.get());
+                }
+                else
+                    mb.append((byte)2); // eog
             }
-            else
-                mb.append((byte)2); // eog
-        }
-        mb.append((byte)0);
-
-        size32_t len = mb.length();
-        Owned<Chunk> freeChunk = getOutOffset(len); // will find space for 'len', might be bigger if from free list
-
-        spillFileIO->write(freeChunk->offset, len, mb.toByteArray());
-
-        savedChunks.enqueue(freeChunk.getClear());
+            mb.append((byte)0);
+            size32_t len = mb.length();
+            chunk.setown(getOutOffset(len)); // will find space for 'len', might be bigger if from free list
+            spillFileIO->write(chunk->offset, len, mb.toByteArray());
 #ifdef TRACE_WRITEAHEAD
-        ActPrintLogEx(&activity->queryContainer(), thorlog_all, MCdebugProgress, "Flushed chunk = %d (savedChunks pos=%d), writeOffset = %"I64F"d, writeSize = %d", inMemRows->queryChunk(), savedChunks.ordinality()-1, freeChunk->offset, len);
+            ActPrintLogEx(&activity->queryContainer(), thorlog_all, MCdebugProgress, "Flushed chunk = %d (savedChunks pos=%d), writeOffset = %"I64F"d, writeSize = %d", inMemRows->queryChunk(), savedChunks.ordinality(), chunk->offset, len);
 #endif
+        }
+
+        savedChunks.enqueue(chunk.getClear());
     }
     virtual size32_t rowSize(const void *row)
     {
@@ -1440,13 +1535,10 @@ ISharedSmartBuffer *createSharedSmartDiskBuffer(CActivityBase *activity, const c
     return new CSharedWriteAheadDisk(activity, spillname, outputs, rowIf, iDiskUsage);
 }
 
-#define MIN_POOL_CHUNKS 10
 class CSharedWriteAheadMem : public CSharedWriteAheadBase
 {
-    QueueOf<CRowSet, false> chunkPool;
     Semaphore poolSem;
     bool writerBlocked;
-    unsigned maxPoolChunks;
 
     virtual void markStop()
     {
@@ -1477,14 +1569,19 @@ class CSharedWriteAheadMem : public CSharedWriteAheadBase
         VALIDATEEQ(queryCOutput(output).currentChunkNum, rowSet->queryChunk());
         return rowSet.getClear();
     }
-    virtual void flushRows()
+    virtual void flushRows(ISharedSmartBufferCallback *callback)
     {
         // NB: called in crit
         if (chunkPool.ordinality() >= maxPoolChunks)
         {
             writerBlocked = true;
-            { CriticalUnblock b(crit);
+            {
+                CriticalUnblock b(crit);
+                if (callback)
+                    callback->blocked();
                 poolSem.wait();
+                if (callback)
+                    callback->unblocked();
                 if (stopped) return;
             }
             unsigned reader=anyReaderBehind();

thorlcr/thorutil/thbuf.hpp

@@ -62,8 +62,16 @@ extern graph_decl ISmartRowBuffer * createSmartInMemoryBuffer(CActivityBase *act
                                                       size32_t buffsize);
 
 // Multiple readers, one writer
+interface ISharedSmartBufferCallback
+{
+    virtual void paged() = 0;
+    virtual void blocked() = 0;
+    virtual void unblocked() = 0;
+};
 interface ISharedSmartBuffer : extends IRowWriter
 {
+    using IRowWriter::putRow;
+    virtual void putRow(const void *row, ISharedSmartBufferCallback *callback) = 0; // extended form of putRow, which signals when pages out via callback
     virtual IRowStream *queryOutput(unsigned output) = 0;
     virtual void cancel()=0;
     virtual void reset() = 0;

thorlcr/thorutil/thormisc.hpp

@@ -62,7 +62,6 @@
 #define THOROPT_PARALLEL_FUNNEL       "parallelFunnel"          // Use parallel funnel impl. if !ordered                                         (default = true)
 #define THOROPT_SORT_MAX_DEVIANCE     "sort_max_deviance"       // Max (byte) variance allowed during sort partitioning                          (default = 10Mb)
 #define THOROPT_OUTPUT_FLUSH_THRESHOLD "output_flush_threshold" // When above limit, workunit result is flushed (committed to Dali)              (default = -1 [off])
-#define THOROPT_OUTPUTLIMIT           "outputLimit"             // OUTPUT Mb limit                                                               (default = 10)
 #define THOROPT_PARALLEL_MATCH        "parallel_match"          // Use multi-threaded join helper (retains sort order without unsorted_output)   (default = false)
 #define THOROPT_UNSORTED_OUTPUT       "unsorted_output"         // Allow Join results to be reodered, implies parallel match                     (default = false)
 #define THOROPT_JOINHELPER_THREADS    "joinHelperThreads"       // Number of threads to use in threaded variety of join helper