HPCC-10674 ECL support for writing JSON files

Signed-off-by: Anthony Fishbeck <anthony.fishbeck@lexisnexis.com>

common/thorhelper/commonext.cpp

@@ -90,6 +90,7 @@ MODULE_INIT(INIT_PRIORITY_STANDARD)
     kindArray[TAKlimit] = "limit";
     kindArray[TAKcsvfetch] = "csvfetch";
     kindArray[TAKxmlwrite] = "xmlwrite";
+    kindArray[TAKjsonwrite] = "jsonwrite";
     kindArray[TAKparse] = "parse";
     kindArray[TAKtopn] = "topn";
     kindArray[TAKmerge] = "merge";

common/thorhelper/thorcommon.cpp

@@ -659,6 +659,7 @@ extern const char * getActivityText(ThorActivityKind kind)
     case TAKlimit:                  return "Limit";
     case TAKcsvfetch:               return "Csv Fetch";
     case TAKxmlwrite:               return "Xml Write";
+    case TAKjsonwrite:              return "Json Write";
     case TAKparse:                  return "Parse";
     case TAKsideeffect:             return "Simple Action";
     case TAKtopn:                   return "Top N";
@@ -839,6 +840,7 @@ extern bool isActivitySink(ThorActivityKind kind)
     case TAKcsvwrite:
     case TAKindexwrite:
     case TAKxmlwrite:
+    case TAKjsonwrite:
     case TAKsoap_rowaction:
     case TAKsoap_datasetaction:
     case TAKkeydiff:

common/thorhelper/thorxmlwrite.cpp

@@ -38,7 +38,7 @@ CommonXmlWriter::~CommonXmlWriter()
     flush(true);
 }
 
-CommonXmlWriter & CommonXmlWriter::clear()
+IXmlWriterExt & CommonXmlWriter::clear()
 {
     out.clear();
     indent = 0;
@@ -361,7 +361,7 @@ CommonJsonWriter::~CommonJsonWriter()
     flush(true);
 }
 
-CommonJsonWriter & CommonJsonWriter::clear()
+IXmlWriterExt & CommonJsonWriter::clear()
 {
     out.clear();
     indent = 0;
@@ -417,6 +417,14 @@ const char *CommonJsonWriter::checkItemNameBeginNested(const char *name)
     return name;
 }
 
+bool CommonJsonWriter::checkUnamedArrayItem(bool begin)
+{
+    CJsonWriterItem *item = (arrays.length()) ? &arrays.tos() : NULL;
+    if (item && item->depth==(begin ? 0 : 1) && item->name.isEmpty())
+        return true;
+    return false;
+}
+
 const char *CommonJsonWriter::checkItemNameEndNested(const char *name)
 {
     CJsonWriterItem *item = (arrays.length()) ? &arrays.tos() : NULL;
@@ -557,13 +565,15 @@ void CommonJsonWriter::outputEndDataset(const char *dsname)
 
 void CommonJsonWriter::outputBeginNested(const char *fieldname, bool nestChildren)
 {
-    if (!fieldname || !*fieldname)
+    if (!fieldname)
+        return;
+    if (!*fieldname && !checkUnamedArrayItem(true))
         return;
 
     flush(false);
     checkFormat(true, false, 1);
     fieldname = checkItemNameBeginNested(fieldname);
-    if (fieldname)
+    if (fieldname && *fieldname)
     {
         const char * sep = (fieldname) ? strchr(fieldname, '/') : NULL;
         while (sep)
@@ -582,13 +592,15 @@ void CommonJsonWriter::outputBeginNested(const char *fieldname, bool nestChildre
 
 void CommonJsonWriter::outputEndNested(const char *fieldname)
 {
-    if (!fieldname || !*fieldname)
+    if (!fieldname)
+        return;
+    if (!*fieldname && !checkUnamedArrayItem(false))
         return;
 
     flush(false);
     checkFormat(false, true, -1);
     fieldname = checkItemNameEndNested(fieldname);
-    if (fieldname)
+    if (fieldname && *fieldname)
     {
         const char * sep = (fieldname) ? strchr(fieldname, '/') : NULL;
         while (sep)
@@ -609,6 +621,37 @@ void CommonJsonWriter::outputSetAll()
     appendJSONValue(out, "All", true);
 }
 
+StringBuffer &buildJsonHeader(StringBuffer  &header, const char *suppliedHeader, const char *rowTag)
+{
+    if (suppliedHeader)
+    {
+        header.append(suppliedHeader);
+        if (rowTag && *rowTag)
+            appendJSONName(header, rowTag).append('[');
+        return header;
+    }
+
+    if (rowTag && *rowTag)
+    {
+        header.append('{');
+        appendJSONName(header, rowTag);
+    }
+
+    return header.append('[');
+}
+
+StringBuffer &buildJsonFooter(StringBuffer  &footer, const char *suppliedFooter, const char *rowTag)
+{
+    if (suppliedFooter)
+    {
+        if (rowTag && *rowTag)
+            footer.append(']');
+        footer.append(suppliedFooter);
+        return footer;
+    }
+    return footer.append((rowTag && *rowTag) ? "]}" : "]");
+}
+
 //=====================================================================================
 
 inline void outputEncodedXmlString(unsigned len, const char *field, const char *fieldname, StringBuffer &out)
@@ -939,7 +982,7 @@ CommonXmlWriter * CreateCommonXmlWriter(unsigned _flags, unsigned _initialIndent
 
 //=====================================================================================
 
-IXmlWriter * createIXmlWriter(unsigned _flags, unsigned _initialIndent, IXmlStreamFlusher *_flusher, XMLWriterType xmlType)
+IXmlWriterExt * createIXmlWriterExt(unsigned _flags, unsigned _initialIndent, IXmlStreamFlusher *_flusher, XMLWriterType xmlType)
 {
     if (xmlType==WTJSON)
         return new CommonJsonWriter(_flags, _initialIndent, _flusher);

common/thorhelper/thorxmlwrite.hpp

@@ -37,17 +37,13 @@ interface IXmlStreamFlusher
     virtual void flushXML(StringBuffer &current, bool isClose) = 0;
 };
 
-class thorhelper_decl CommonXmlWriter : public CInterface, implements IXmlWriter
+class thorhelper_decl CommonXmlWriter : public CInterface, implements IXmlWriterExt
 {
 public:
     CommonXmlWriter(unsigned _flags, unsigned initialIndent=0,  IXmlStreamFlusher *_flusher=NULL);
     ~CommonXmlWriter();
     IMPLEMENT_IINTERFACE;
 
-    CommonXmlWriter & clear();
-    unsigned length() const                                 { return out.length(); }
-    const char * str() const                                { return out.str(); }
-
     void outputBeginNested(const char *fieldname, bool nestChildren, bool doIndent);
     void outputEndNested(const char *fieldname, bool doIndent);
 
@@ -73,6 +69,11 @@ public:
     virtual void outputSetAll();
     virtual void outputXmlns(const char *name, const char *uri);
 
+    //IXmlWriterExt
+    virtual IXmlWriterExt & clear();
+    virtual unsigned length() const                                 { return out.length(); }
+    virtual const char * str() const                                { return out.str(); }
+
 protected:
     bool checkForAttribute(const char * fieldname);
     void closeTag();
@@ -91,16 +92,13 @@ protected:
     bool tagClosed;
 };
 
-class thorhelper_decl CommonJsonWriter : public CInterface, implements IXmlWriter
+class thorhelper_decl CommonJsonWriter : public CInterface, implements IXmlWriterExt
 {
 public:
     CommonJsonWriter(unsigned _flags, unsigned initialIndent=0,  IXmlStreamFlusher *_flusher=NULL);
     ~CommonJsonWriter();
     IMPLEMENT_IINTERFACE;
 
-    CommonJsonWriter & clear();
-    unsigned length() const                                 { return out.length(); }
-    const char * str() const                                { return out.str(); }
     void checkDelimit(int inc=0);
     void checkFormat(bool doDelimit, bool needDelimiter=true, int inc=0);
 
@@ -130,6 +128,11 @@ public:
     virtual void outputSetAll();
     virtual void outputXmlns(const char *name, const char *uri){}
 
+    //IXmlWriterExt
+    virtual IXmlWriterExt & clear();
+    virtual unsigned length() const                                 { return out.length(); }
+    virtual const char * str() const                                { return out.str(); }
+
     void outputBeginRoot(){out.append('{');}
     void outputEndRoot(){out.append('}');}
 
@@ -153,6 +156,8 @@ protected:
     const char *checkItemName(const char *name, bool simpleType=true);
     const char *checkItemNameBeginNested(const char *name);
     const char *checkItemNameEndNested(const char *name);
+    bool checkUnamedArrayItem(bool begin);
+
 
     IXmlStreamFlusher *flusher;
     CIArrayOf<CJsonWriterItem> arrays;
@@ -163,6 +168,9 @@ protected:
     bool needDelimiter;
 };
 
+thorhelper_decl StringBuffer &buildJsonHeader(StringBuffer  &header, const char *suppliedHeader, const char *rowTag);
+thorhelper_decl StringBuffer &buildJsonFooter(StringBuffer  &footer, const char *suppliedFooter, const char *rowTag);
+
 //Writes type encoded XML strings  (xsi:type="xsd:string", xsi:type="xsd:boolean" etc)
 class thorhelper_decl CommonEncodedXmlWriter : public CommonXmlWriter
 {
@@ -192,7 +200,7 @@ public:
 
 enum XMLWriterType{WTStandard, WTEncoding, WTEncodingData64, WTJSON} ;
 thorhelper_decl CommonXmlWriter * CreateCommonXmlWriter(unsigned _flags, unsigned initialIndent=0, IXmlStreamFlusher *_flusher=NULL, XMLWriterType xmlType=WTStandard);
-thorhelper_decl IXmlWriter * createIXmlWriter(unsigned _flags, unsigned initialIndent=0, IXmlStreamFlusher *_flusher=NULL, XMLWriterType xmlType=WTStandard);
+thorhelper_decl IXmlWriterExt * createIXmlWriterExt(unsigned _flags, unsigned initialIndent=0, IXmlStreamFlusher *_flusher=NULL, XMLWriterType xmlType=WTStandard);
 
 class thorhelper_decl SimpleOutputWriter : public CInterface, implements IXmlWriter
 {

common/workunit/referencedfilelist.cpp

@@ -610,7 +610,7 @@ void ReferencedFileList::addFilesFromQuery(IConstWorkUnit *cw, const IHpccPackag
                 continue;
             ThorActivityKind kind = (ThorActivityKind) node.getPropInt("att[@name='_kind']/@value", TAKnone);
             //not likely to be part of roxie queries, but for forward compatibility:
-            if(kind==TAKdiskwrite || kind==TAKindexwrite || kind==TAKcsvwrite || kind==TAKxmlwrite)
+            if(kind==TAKdiskwrite || kind==TAKindexwrite || kind==TAKcsvwrite || kind==TAKxmlwrite || kind==TAKjsonwrite)
                 continue;
             if (node.getPropBool("att[@name='_isSpill']/@value") ||
                 node.getPropBool("att[@name='_isTransformSpill']/@value"))

ecl/eclagent/eclgraph.cpp

@@ -140,6 +140,7 @@ static IHThorActivity * createActivity(IAgentContext & agent, unsigned activityI
     case TAKcsvwrite:
         return createCsvWriteActivity(agent, activityId, subgraphId, (IHThorCsvWriteArg &)arg, kind);
     case TAKxmlwrite:
+    case TAKjsonwrite:
         return createXmlWriteActivity(agent, activityId, subgraphId, (IHThorXmlWriteArg &)arg, kind);
     case TAKpipethrough:
         return createPipeThroughActivity(agent, activityId, subgraphId, (IHThorPipeThroughArg &)arg, kind);
@@ -371,6 +372,7 @@ bool EclGraphElement::alreadyUpToDate(IAgentContext & agent)
     case TAKdiskwrite:
     case TAKcsvwrite:
     case TAKxmlwrite:
+    case TAKjsonwrite:
         {
             IHThorDiskWriteArg * helper = static_cast<IHThorDiskWriteArg *>(arg.get());
             filename.set(helper->getFileName());
@@ -578,6 +580,7 @@ bool EclGraphElement::prepare(IAgentContext & agent, const byte * parentExtract,
         case TAKdiskwrite:
         case TAKcsvwrite:
         case TAKxmlwrite:
+        case TAKjsonwrite:
             alreadyUpdated = alreadyUpToDate(agent);
             if (alreadyUpdated)
                 return false;

ecl/hql/hqlatoms.cpp

@@ -215,6 +215,7 @@ IAtom * isNullAtom;
 IAtom * isValidAtom;
 IAtom * jobAtom;
 IAtom * jobTempAtom;
+IAtom * jsonAtom;
 IAtom * keepAtom;
 IAtom * keyedAtom;
 IAtom * labeledAtom;
@@ -634,6 +635,7 @@ MODULE_INIT(INIT_PRIORITY_HQLATOM)
     MAKEATOM(isValid);
     MAKEATOM(job);
     MAKEATOM(jobTemp);
+    MAKEATOM(json);
     MAKEATOM(keep);
     MAKEATOM(keyed);
     MAKEATOM(labeled);

ecl/hql/hqlatoms.hpp

@@ -218,6 +218,7 @@ extern HQL_API IAtom * isNullAtom;
 extern HQL_API IAtom * isValidAtom;
 extern HQL_API IAtom * jobAtom;
 extern HQL_API IAtom * jobTempAtom;
+extern HQL_API IAtom * jsonAtom;
 extern HQL_API IAtom * keepAtom;
 extern HQL_API IAtom * keyedAtom;
 extern HQL_API IAtom * labeledAtom;

ecl/hql/hqlgram.y

@@ -256,6 +256,7 @@ static void eclsyntaxerror(HqlGram * parser, const char * s, short yystate, int
   ITERATE
   JOIN
   JOINED
+  JSON_TOKEN
   KEEP
   KEYDIFF
   KEYED
@@ -3212,6 +3213,16 @@ outputFlag
                             $3.unwindCommaList(args);
                             $$.setExpr(createExprAttribute(xmlAtom, args), $1);
                         }
+    | JSON_TOKEN        {
+                            $$.setExpr(createAttribute(jsonAtom));
+                            $$.setPosition($1);
+                        }
+    | JSON_TOKEN '(' xmlOptions ')' //exact same options as XML for now
+                        {
+                            HqlExprArray args;
+                            $3.unwindCommaList(args);
+                            $$.setExpr(createExprAttribute(jsonAtom, args), $1);
+                        }
     | UPDATE            {
                             $$.setExpr(createComma(createAttribute(updateAtom), createAttribute(overwriteAtom)), $1);
                         }

ecl/hql/hqlgram2.cpp

@@ -10401,6 +10401,7 @@ static void getTokenText(StringBuffer & msg, int token)
     case ITERATE: msg.append("ITERATE"); break;
     case JOIN: msg.append("JOIN"); break;
     case JOINED: msg.append("JOINED"); break;
+    case JSON_TOKEN: msg.append("JSON"); break;
     case KEEP: msg.append("KEEP"); break;
     case KEYDIFF: msg.append("KEYDIFF"); break;
     case KEYED: msg.append("KEYED"); break;

ecl/hql/hqllex.l

@@ -738,6 +738,7 @@ ISVALID             { RETURNSYM(ISVALID); }
 ITERATE             { RETURNSYM(ITERATE); }
 JOIN                { RETURNSYM(JOIN); }
 JOINED              { RETURNSYM(JOINED); }
+JSON                { RETURNSYM(JSON_TOKEN); }
 KEEP                { RETURNSYM(KEEP); }
 KEYDIFF             { RETURNSYM(KEYDIFF); }
 KEYED               { RETURNSYM(KEYED); }

ecl/hqlcpp/hqlgraph.cpp

@@ -658,6 +658,8 @@ const char * LogicalGraphCreator::getActivityText(IHqlExpression * expr, StringB
                 temp.append("Output");
                 if (expr->hasAttribute(xmlAtom))
                     temp.append(" XML");
+                if (expr->hasAttribute(jsonAtom))
+                    temp.append(" JSON");
                 else if (expr->hasAttribute(csvAtom))
                     temp.append(" CSV");
                 queryExpandFilename(temp, filename);

ecl/hqlcpp/hqlhtcpp.cpp

@@ -10336,7 +10336,14 @@ ABoundActivity * HqlCppTranslator::doBuildActivityOutput(BuildCtx & ctx, IHqlExp
     OwnedHqlExpr filename = foldHqlExpression(rawFilename);
     IHqlExpression * program  = queryRealChild(expr, 2);
     IHqlExpression * csvAttr = expr->queryAttribute(csvAtom);
+    bool isJson = false;
     IHqlExpression * xmlAttr = expr->queryAttribute(xmlAtom);
+    if (!xmlAttr)
+    {
+        xmlAttr = expr->queryAttribute(jsonAtom);
+        if (xmlAttr)
+            isJson=true;
+    }
     LinkedHqlExpr expireAttr = expr->queryAttribute(expireAtom);
     IHqlExpression * seq = querySequence(expr);
 
@@ -10354,30 +10361,30 @@ ABoundActivity * HqlCppTranslator::doBuildActivityOutput(BuildCtx & ctx, IHqlExp
 
     Owned<ABoundActivity> boundDataset = buildCachedActivity(ctx, dataset);
     ThorActivityKind kind = TAKdiskwrite;
-    const char * activity = "DiskWrite";
+    const char * activityArgName = "DiskWrite";
     if (expr->getOperator() == no_spill)
     {
         kind = TAKspill;
-        activity = "Spill";
+        activityArgName = "Spill";
     }
     else if (pipe)
     {
         kind = TAKpipewrite;
-        activity = "PipeWrite";
+        activityArgName = "PipeWrite";
     }
     else if (csvAttr)
     {
         kind = TAKcsvwrite;
-        activity = "CsvWrite";
+        activityArgName = "CsvWrite";
     }
     else if (xmlAttr)
     {
-        kind = TAKxmlwrite;
-        activity = "XmlWrite";
+        kind = (isJson) ? TAKjsonwrite : TAKxmlwrite;
+        activityArgName = "XmlWrite";
     }
 
     bool useImplementationClass = options.minimizeActivityClasses && targetRoxie() && expr->hasAttribute(_spill_Atom);
-    Owned<ActivityInstance> instance = new ActivityInstance(*this, ctx, kind, expr, activity);
+    Owned<ActivityInstance> instance = new ActivityInstance(*this, ctx, kind, expr, activityArgName);
     //Output to a variable filename is either a user result, or a computed workflow spill, both need evaluating.
 
     if (useImplementationClass)
@@ -10535,7 +10542,8 @@ ABoundActivity * HqlCppTranslator::doBuildActivityOutput(BuildCtx & ctx, IHqlExp
         OwnedHqlExpr outputDs = createDataset(no_null, LINK(outputRecord));
         HqlExprArray xmlnsAttrs;
         gatherAttributes(xmlnsAttrs, xmlnsAtom, expr);
-        Owned<IWUResult> result = createDatasetResultSchema(seq, queryResultName(expr), outputRecord, xmlnsAttrs, (kind != TAKcsvwrite) && (kind != TAKxmlwrite), true);
+        bool createTransformer = (kind != TAKcsvwrite) && (kind != TAKxmlwrite) && (kind != TAKjsonwrite);
+        Owned<IWUResult> result = createDatasetResultSchema(seq, queryResultName(expr), outputRecord, xmlnsAttrs, createTransformer, true);
         if (expr->hasAttribute(resultAtom))
             result->setResultRowLimit(-1);
 

ecl/hqlcpp/hqlresource.cpp

@@ -3224,7 +3224,7 @@ void EclResourcer::createInitialGraph(IHqlExpression * expr, IHqlExpression * ow
                 //Needs the grouping to be saved in the same way.  Could cope with compressed matching, but not
                 //much point - since fairly unlikely.
                 IHqlExpression * filename = expr->queryChild(1);
-                if (filename && (filename->getOperator() == no_constant) && !expr->hasAttribute(xmlAtom) && !expr->hasAttribute(csvAtom))
+                if (filename && (filename->getOperator() == no_constant) && !expr->hasAttribute(xmlAtom) && !expr->hasAttribute(jsonAtom) && !expr->hasAttribute(csvAtom))
                 {
                     IHqlExpression * dataset = expr->queryChild(0);
                     if (expr->hasAttribute(groupedAtom) == isGrouped(dataset))

ecl/hthor/hthor.cpp

@@ -880,7 +880,7 @@ void CHThorCsvWriteActivity::setFormat(IFileDescriptor * desc)
 
 //=====================================================================================================
 
-CHThorXmlWriteActivity::CHThorXmlWriteActivity(IAgentContext &_agent, unsigned _activityId, unsigned _subgraphId, IHThorXmlWriteArg &_arg, ThorActivityKind _kind) : CHThorDiskWriteActivity(_agent, _activityId, _subgraphId, _arg, _kind), helper(_arg)
+CHThorXmlWriteActivity::CHThorXmlWriteActivity(IAgentContext &_agent, unsigned _activityId, unsigned _subgraphId, IHThorXmlWriteArg &_arg, ThorActivityKind _kind) : CHThorDiskWriteActivity(_agent, _activityId, _subgraphId, _arg, _kind), helper(_arg), headerLength(0), footerLength(0)
 {
     OwnedRoxieString xmlpath(helper.getXmlIteratorPath());
     if (!xmlpath)
@@ -898,12 +898,23 @@ void CHThorXmlWriteActivity::execute()
 {
     // Loop thru the results
     numRecords = 0;
+    StringBuffer header;
     OwnedRoxieString suppliedHeader(helper.getHeader());
-    const char *header = suppliedHeader;
-    if (!header) header = "<Dataset>\n";
-    diskout->write(strlen(header), header);
+    if (kind==TAKjsonwrite)
+    {
+        buildJsonHeader(header, suppliedHeader, rowTag);
+        headerLength = header.length();
+    }
+    else if (suppliedHeader)
+        header.set(suppliedHeader);
+    else
+        header.set("<Dataset>\n");
+    diskout->write(header.length(), header.str());
+
+    Owned<IXmlWriterExt> writer = createIXmlWriterExt(helper.getXmlFlags(), 0, NULL, (kind==TAKjsonwrite) ? WTJSON : WTStandard);
+    writer->outputBeginArray(rowTag); //need to set up the array
+    writer->clear(); //but not output it
 
-    CommonXmlWriter xmlOutput(helper.getXmlFlags());
     loop
     {
         OwnedConstRoxieRow nextrec(input->nextInGroup());
@@ -916,29 +927,44 @@ void CHThorXmlWriteActivity::execute()
 
         try
         {
-            xmlOutput.clear().outputBeginNested(rowTag, false);
-            helper.toXML((const byte *)nextrec.get(), xmlOutput);
-            xmlOutput.outputEndNested(rowTag);
+            writer->clear().outputBeginNested(rowTag, false);
+            helper.toXML((const byte *)nextrec.get(), *writer);
+            writer->outputEndNested(rowTag);
         }
         catch(IException * e)
         {
             throw makeWrappedException(e);
         }
 
-        diskout->write(xmlOutput.length(), xmlOutput.str());
+        diskout->write(writer->length(), writer->str());
         numRecords++;
     }
+
     OwnedRoxieString suppliedFooter(helper.getFooter());
-    const char *footer = suppliedFooter;
-    if (!footer) footer = "</Dataset>\n";
-    diskout->write(strlen(footer), footer);
+    StringBuffer footer;
+    if (kind==TAKjsonwrite)
+    {
+        buildJsonFooter(footer.newline(), suppliedFooter, rowTag);
+        footerLength=footer.length()-1;
+    }
+    else if (suppliedFooter)
+        footer.append(suppliedFooter);
+    else
+        footer.append("</Dataset>");
+
+    diskout->write(footer.length(), footer);
 }
 
 void CHThorXmlWriteActivity::setFormat(IFileDescriptor * desc)
 {
     desc->queryProperties().setProp("@format","utf8n");
     desc->queryProperties().setProp("@rowTag",rowTag.str());
-    desc->queryProperties().setProp("@kind", "xml");
+    desc->queryProperties().setProp("@kind", (kind==TAKjsonwrite) ? "json" : "xml");
+    if (headerLength)
+        desc->queryProperties().setPropInt("@headerLength", headerLength);
+    if (footerLength)
+        desc->queryProperties().setPropInt("@footerLength", footerLength);
+
     const char *recordECL = helper.queryRecordECL();
     if (recordECL && *recordECL)
         desc->queryProperties().setProp("ECL", recordECL);

ecl/hthor/hthor.ipp

@@ -348,6 +348,8 @@ class CHThorXmlWriteActivity : public CHThorDiskWriteActivity
 {
     IHThorXmlWriteArg &helper;
     StringBuffer rowTag;
+    unsigned headerLength;
+    unsigned footerLength;
 
     virtual bool isOutputTransformed() { return true; }
     virtual void setFormat(IFileDescriptor * desc);

esp/services/ws_workunits/ws_workunitsQuerySets.cpp

@@ -190,7 +190,7 @@ bool copyWULogicalFiles(IEspContext &context, IConstWorkUnit &cw, const char *cl
                 IPropertyTree &node = iter->query();
                 ThorActivityKind kind = (ThorActivityKind) node.getPropInt("att[@name='_kind']/@value", TAKnone);
 
-                if(kind==TAKdiskwrite || kind==TAKindexwrite || kind==TAKcsvwrite || kind==TAKxmlwrite)
+                if(kind==TAKdiskwrite || kind==TAKindexwrite || kind==TAKcsvwrite || kind==TAKxmlwrite || kind==TAKjsonwrite)
                     continue;
                 if (node.getPropBool("att[@name='_isSpill']/@value") || node.getPropBool("att[@name='_isTransformSpill']/@value"))
                     continue;

plugins/CMakeLists.txt

@@ -28,4 +28,3 @@ add_subdirectory (v8embed)
 add_subdirectory (pyembed)
 add_subdirectory (javaembed)
 add_subdirectory (Rembed)
-add_subdirectory (cassandra)

roxie/ccd/ccdquery.cpp

@@ -561,6 +561,7 @@ protected:
         case TAKcsvwrite:
         case TAKdiskwrite:
         case TAKxmlwrite:
+        case TAKjsonwrite:
         case TAKmemoryspillwrite:
             return createRoxieServerDiskWriteActivityFactory(id, subgraphId, *this, helperFactory, kind, isRootAction(node));
         case TAKindexwrite:

roxie/ccd/ccdserver.cpp

@@ -11106,10 +11106,13 @@ class CRoxieServerXmlWriteActivity : public CRoxieServerDiskWriteActivity
 {
     IHThorXmlWriteArg &xmlHelper;
     StringAttr rowTag;
+    ThorActivityKind kind;
+    unsigned headerLength;
+    unsigned footerLength;
 
 public:
-    CRoxieServerXmlWriteActivity(const IRoxieServerActivityFactory *_factory, IProbeManager *_probeManager)
-        : CRoxieServerDiskWriteActivity(_factory, _probeManager), xmlHelper(static_cast<IHThorXmlWriteArg &>(helper))
+    CRoxieServerXmlWriteActivity(const IRoxieServerActivityFactory *_factory, IProbeManager *_probeManager, ThorActivityKind _kind)
+        : CRoxieServerDiskWriteActivity(_factory, _probeManager), xmlHelper(static_cast<IHThorXmlWriteArg &>(helper)), kind(_kind), headerLength(0), footerLength(0)
     {
     }
 
@@ -11130,11 +11133,23 @@ public:
 
     virtual void onExecute() 
     {
+        StringBuffer header;
         OwnedRoxieString suppliedHeader(xmlHelper.getHeader());
-        const char *header = suppliedHeader;
-        if (!header) header = "<Dataset>\n";
-        diskout->write(strlen(header), header);
-        CommonXmlWriter xmlOutput(xmlHelper.getXmlFlags());
+        if (kind==TAKjsonwrite)
+        {
+            buildJsonHeader(header, suppliedHeader, rowTag);
+            headerLength = header.length();
+        }
+        else if (suppliedHeader)
+            header.set(suppliedHeader);
+        else
+            header.set("<Dataset>\n");
+        diskout->write(header.length(), header.str());
+
+        Owned<IXmlWriterExt> writer = createIXmlWriterExt(xmlHelper.getXmlFlags(), 0, NULL, (kind==TAKjsonwrite) ? WTJSON : WTStandard);
+        writer->outputBeginArray(rowTag); //need to set this
+        writer->clear(); //but not output it
+
         loop
         {
             OwnedConstRoxieRow nextrec = input->nextInGroup();
@@ -11145,15 +11160,24 @@ public:
                     break;
             }
             processed++;
-            xmlOutput.clear().outputBeginNested(rowTag, false);
-            xmlHelper.toXML((const byte *)nextrec.get(), xmlOutput);
-            xmlOutput.outputEndNested(rowTag);
-            diskout->write(xmlOutput.length(), xmlOutput.str());
+            writer->clear().outputBeginNested(rowTag, false);
+            xmlHelper.toXML((const byte *)nextrec.get(), *writer);
+            writer->outputEndNested(rowTag);
+            diskout->write(writer->length(), writer->str());
         }
         OwnedRoxieString suppliedFooter(xmlHelper.getFooter());
-        const char * footer = suppliedFooter;
-        if (!footer) footer = "</Dataset>\n";
-        diskout->write(strlen(footer), footer);
+        StringBuffer footer;
+        if (kind==TAKjsonwrite)
+        {
+            buildJsonFooter(footer.newline(), suppliedFooter, rowTag);
+            footerLength=footer.length()-1;
+        }
+        else if (suppliedFooter)
+            footer.append(suppliedFooter);
+        else
+            footer.append("</Dataset>");
+
+        diskout->write(footer.length(), footer);
     }
 
     virtual void reset()
@@ -11167,7 +11191,11 @@ public:
         CRoxieServerDiskWriteActivity::setFileProperties(desc);
         desc->queryProperties().setProp("@format","utf8n");
         desc->queryProperties().setProp("@rowTag",rowTag.get());
-        desc->queryProperties().setProp("@kind", "xml");
+        desc->queryProperties().setProp("@kind", (kind==TAKjsonwrite) ? "json" : "xml");
+        if (headerLength)
+            desc->queryProperties().setPropInt("@headerLength", headerLength);
+        if (footerLength)
+            desc->queryProperties().setPropInt("@footerLength", footerLength);
     }
 
     virtual bool isOutputTransformed() const { return true; }
@@ -11197,7 +11225,9 @@ public:
             {
             case TAKdiskwrite: return new CRoxieServerDiskWriteActivity(this, _probeManager);
             case TAKcsvwrite: return new CRoxieServerCsvWriteActivity(this, _probeManager);
-            case TAKxmlwrite: return new CRoxieServerXmlWriteActivity(this, _probeManager);
+            case TAKxmlwrite:
+            case TAKjsonwrite:
+                return new CRoxieServerXmlWriteActivity(this, _probeManager, kind);
             };
             throwUnexpected();
         case 1:
@@ -19822,7 +19852,7 @@ public:
                     unsigned int writeFlags = serverContext->getXmlFlags();
                     if (response->mlFmt==MarkupFmt_JSON)
                         writeFlags |= XWFnoindent;
-                    writer.setown(createIXmlWriter(writeFlags, 1, response, (response->mlFmt==MarkupFmt_JSON) ? WTJSON : WTStandard));
+                    writer.setown(createIXmlWriterExt(writeFlags, 1, response, (response->mlFmt==MarkupFmt_JSON) ? WTJSON : WTStandard));
                     writer->outputBeginArray("Row");
                 }
             }

rtl/include/eclhelper.hpp

@@ -181,6 +181,12 @@ public:
     inline void outputCString(const char *field, const char *fieldname) { outputString((size32_t)strlen(field), field, fieldname); }
 };
 
+interface IXmlWriterExt : extends IXmlWriter
+{
+    virtual IXmlWriterExt & clear() = 0;
+    virtual unsigned length() const = 0;
+    virtual const char *str() const = 0;
+};
 
 interface IFieldProcessor : public IInterface
 {
@@ -919,6 +925,7 @@ enum ThorActivityKind
     TAKunknowndenormalizegroup2,
     TAKunknowndenormalizegroup3,
     TAKlastdenormalizegroup,
+    TAKjsonwrite,
 
     TAKlast
 };

testing/regress/ecl/jsonout.ecl

@@ -0,0 +1,62 @@
+/*##############################################################################
+
+    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.
+############################################################################## */
+
+phoneRecord :=
+            RECORD
+string5         areaCode{xpath('areaCode')};
+udecimal12      number{xpath('number')};
+            END;
+
+contactrecord :=
+            RECORD
+phoneRecord     phone;
+boolean         hasemail{xpath('hasEmail')};
+                ifblock(self.hasemail)
+string              email;
+                end;
+            END;
+
+bookRec :=
+    RECORD
+string      title;
+string      author;
+    END;
+
+personRecord :=
+            RECORD
+string20        surname;
+string10        forename;
+phoneRecord     homePhone;
+boolean         hasMobile;
+                ifblock(self.hasMobile)
+phoneRecord         mobilePhone;
+                end;
+contactRecord   contact;
+dataset(bookRec) books;
+set of string    colours;
+string2         endmarker := '$$';
+            END;
+
+namesTable := dataset([
+        {'Halliday','Gavin','09876',123456,true,'07967',838690, 'n/a','n/a',true,'gavin@edata.com',[{'To kill a mocking bird','Lee'},{'Zen and the art of motorcycle maintainence','Pirsig'}], ALL},
+        {'Halliday','Abigail','09876',654321,false,'','',false,[{'The cat in the hat','Suess'},{'Wolly the sheep',''}], ['Red','Yellow']}
+        ], personRecord);
+
+output(namesTable,,'REGRESS::TEMP::output.json',overwrite, json('jrow'));
+
+//will read back data and add workunit output when json read is implemented
+

testing/regress/ecl/key/jsonout.xml

@@ -0,0 +1,2 @@
+<Dataset name='Result 1'>
+</Dataset>

thorlcr/activities/xmlwrite/thxmlwrite.cpp

@@ -27,9 +27,12 @@
 class CXmlWriteActivityMaster : public CWriteMasterBase
 {
     IHThorXmlWriteArg *helper;
+    ThorActivityKind kind;
+    unsigned headerLength;
+    unsigned footerLength;
 
 public:
-    CXmlWriteActivityMaster(CMasterGraphElement *info) : CWriteMasterBase(info)
+    CXmlWriteActivityMaster(CMasterGraphElement *info, ThorActivityKind _kind) : CWriteMasterBase(info), kind(_kind), headerLength(0), footerLength(0)
     {
         helper = (IHThorXmlWriteArg *)queryHelper();
     }
@@ -52,12 +55,19 @@ public:
         }
         props.setProp("@rowTag", rowTag.str());
         props.setProp("@format", "utf8n");
-        props.setProp("@kind", "xml");
+        props.setProp("@kind", (kind==TAKjsonwrite) ? "json" : "xml");
+
+        StringBuffer s;
+        OwnedRoxieString supplied(helper->getHeader());
+        props.setPropInt("@headerLength", buildJsonHeader(s, supplied, rowTag).length());
+
+        supplied.set(helper->getFooter());
+        props.setPropInt("@footerLength", buildJsonFooter(s.clear(), supplied, rowTag).length());
     }
 };
 
-CActivityBase *createXmlWriteActivityMaster(CMasterGraphElement *container)
+CActivityBase *createXmlWriteActivityMaster(CMasterGraphElement *container, ThorActivityKind kind)
 {
-    return new CXmlWriteActivityMaster(container);
+    return new CXmlWriteActivityMaster(container, kind);
 }
 

thorlcr/activities/xmlwrite/thxmlwrite.ipp

@@ -22,7 +22,7 @@
 #include "thdiskbase.ipp"
 
 
-CActivityBase *createXmlWriteActivityMaster(CMasterGraphElement *container);
+CActivityBase *createXmlWriteActivityMaster(CMasterGraphElement *container, ThorActivityKind kind);
 
 
 #endif

thorlcr/activities/xmlwrite/thxmlwriteslave.cpp

@@ -31,9 +31,10 @@
 class CXmlWriteSlaveActivity : public CDiskWriteSlaveActivityBase
 {
     IHThorXmlWriteArg *helper;
+    ThorActivityKind kind;
 
 public:
-    CXmlWriteSlaveActivity(CGraphElementBase *container) : CDiskWriteSlaveActivityBase(container)
+    CXmlWriteSlaveActivity(CGraphElementBase *container, ThorActivityKind _kind) : CDiskWriteSlaveActivityBase(container), kind(_kind)
     {
         helper = static_cast <IHThorXmlWriteArg *> (queryHelper());
     }
@@ -53,50 +54,55 @@ public:
             rowTag.append(path);
         }
 
-        StringBuffer xmlOutput;
-        CommonXmlWriter xmlWriter(helper->getXmlFlags());
+        StringBuffer out;
         if (!dlfn.isExternal() || firstNode()) // if external, 1 header,footer
         {
-            OwnedRoxieString header(helper->getHeader());
-            if (header)
-                xmlOutput.clear().append(header);
+            OwnedRoxieString suppliedHeader(helper->getHeader());
+            if (kind==TAKjsonwrite)
+                buildJsonHeader(out, suppliedHeader, rowTag);
+            else if (suppliedHeader)
+                out.set(suppliedHeader);
             else
-                xmlOutput.clear().append("<Dataset>").newline();
-            outraw->write(xmlOutput.length(), xmlOutput.toCharArray());
+                out.set("<Dataset>").newline();
+            outraw->write(out.length(), out.toCharArray());
             if (calcFileCrc)
-                fileCRC.tally(xmlOutput.length(), xmlOutput.toCharArray());
+                fileCRC.tally(out.length(), out.toCharArray());
         }
+        Owned<IXmlWriterExt> writer = createIXmlWriterExt(helper->getXmlFlags(), 0, NULL, (kind==TAKjsonwrite) ? WTJSON : WTStandard);
+        writer->outputBeginArray(rowTag); //need this to format rows, even if not outputting it below
         while(!abortSoon)
         {
             OwnedConstThorRow row = input->ungroupedNextRow();
             if (!row)
                 break;
-            xmlWriter.clear().outputBeginNested(rowTag, false);
-            helper->toXML((const byte *)row.get(), xmlWriter);
-            xmlWriter.outputEndNested(rowTag);
-            outraw->write(xmlWriter.length(), xmlWriter.str());
+            writer->clear().outputBeginNested(rowTag, false);
+            helper->toXML((const byte *)row.get(), *writer);
+            writer->outputEndNested(rowTag);
+            outraw->write(writer->length(), writer->str());
             if (calcFileCrc)
-                fileCRC.tally(xmlWriter.length(), xmlWriter.str());
+                fileCRC.tally(writer->length(), writer->str());
             processed++;
         }
         if (!dlfn.isExternal() || lastNode()) // if external, 1 header,footer
         {
-            OwnedRoxieString footer(helper->getFooter());
-            if (footer)
-                xmlOutput.clear().append(footer);
+            OwnedRoxieString suppliedFooter(helper->getFooter());
+            if (kind==TAKjsonwrite)
+                buildJsonFooter(out.clear().newline(), suppliedFooter, rowTag);
+            else if (suppliedFooter)
+                out.set(suppliedFooter);
             else
-                xmlOutput.clear().append("</Dataset>").newline();
-            outraw->write(xmlOutput.length(), xmlOutput.toCharArray());
+                out.set("</Dataset>").newline();
+            outraw->write(out.length(), out.toCharArray());
             if (calcFileCrc)
-                fileCRC.tally(xmlOutput.length(), xmlOutput.toCharArray());
+                fileCRC.tally(out.length(), out.toCharArray());
         }
     }
     virtual bool wantRaw() { return true; }
 };
 
-CActivityBase *createXmlWriteSlave(CGraphElementBase *container)
+CActivityBase *createXmlWriteSlave(CGraphElementBase *container, ThorActivityKind kind)
 {
-    return new CXmlWriteSlaveActivity(container);
+    return new CXmlWriteSlaveActivity(container, kind);
 }
 
 

thorlcr/activities/xmlwrite/thxmlwriteslave.ipp

@@ -21,6 +21,6 @@
 #include "slave.ipp"
 #include "thdiskbaseslave.ipp"
 
-activityslaves_decl CActivityBase *createXmlWriteSlave(CGraphElementBase *container);
+activityslaves_decl CActivityBase *createXmlWriteSlave(CGraphElementBase *container, ThorActivityKind kind);
 
 #endif

thorlcr/graph/thgraph.cpp

@@ -620,6 +620,7 @@ bool CGraphElementBase::prepareContext(size32_t parentExtractSz, const byte *par
             case TAKdiskwrite:
             case TAKcsvwrite:
             case TAKxmlwrite:
+            case TAKjsonwrite:
                 if (_shortCircuit) return true;
                 onCreate();
                 alreadyUpdated = checkUpdate();
@@ -885,6 +886,7 @@ bool isGlobalActivity(CGraphElementBase &container)
 // always global, but only co-ordinate init/done
         case TAKcsvwrite:
         case TAKxmlwrite:
+        case TAKjsonwrite:
         case TAKindexwrite:
         case TAKkeydiff:
         case TAKkeypatch:

thorlcr/graph/thgraphmaster.cpp

@@ -551,6 +551,7 @@ bool CMasterGraphElement::checkUpdate()
         case TAKdiskwrite:
         case TAKcsvwrite:
         case TAKxmlwrite:
+        case TAKjsonwrite:
         {
             IHThorDiskWriteArg *helper = (IHThorDiskWriteArg *)queryHelper();
             doCheckUpdate = 0 != (helper->getFlags() & TDWupdate);

thorlcr/master/thactivitymaster.cpp

@@ -343,7 +343,8 @@ public:
                 ret = createXmlReadActivityMaster(this);
                 break;
             case TAKxmlwrite:
-                ret = createXmlWriteActivityMaster(this);
+            case TAKjsonwrite:
+                ret = createXmlWriteActivityMaster(this, kind);
                 break;
             case TAKmerge:
                 ret = createMergeActivityMaster(this);

thorlcr/slave/slave.cpp

@@ -662,7 +662,8 @@ public:
                 ret = createXmlReadSlave(this);
                 break;
             case TAKxmlwrite:
-                ret = createXmlWriteSlave(this);
+            case TAKjsonwrite:
+                ret = createXmlWriteSlave(this, kind);
                 break;
             case TAKmerge:
                 if (queryLocalOrGrouped())