HPCC-10455 Support streamed dataset results from embedded languages

Initial POC implementation in Python plugin

Also fixing HPCC-10457 Streamed dataset support for Python

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

ecl/hqlcpp/hqltcppc2.cpp

@@ -278,7 +278,7 @@ void CChildDatasetColumnInfo::buildDeserialize(HqlCppTranslator & translator, Bu
         if (serializeForm == diskAtom)
         {
             //If we ever generate the meta definition for an internal serialization format then the following needs to be implemented
-            UNIMPLEMENTED_X("deserialize serialized dictionary from disk");
+            // UNIMPLEMENTED_X("deserialize serialized dictionary from disk");
             return;
         }
     }
@@ -314,7 +314,7 @@ void CChildDatasetColumnInfo::buildSerialize(HqlCppTranslator & translator, Buil
         if (serializeForm == diskAtom)
         {
             //If we ever generate the meta definition for an internal serialization format then the following needs to be implemented
-            UNIMPLEMENTED_X("deserialize serialized dictionary from disk");
+            return; //UNIMPLEMENTED_X("deserialize serialized dictionary from disk");
         }
     }
 

plugins/Rembed/Rembed.cpp

@@ -366,6 +366,11 @@ public:
         }
     }
 
+    virtual IRowStream *getDatasetResult(IEngineRowAllocator * _resultAllocator)
+    {
+        UNIMPLEMENTED;
+    }
+
     virtual void bindBooleanParam(const char *name, bool val)
     {
         R[name] = val;

plugins/javaembed/javaembed.cpp

@@ -723,7 +723,10 @@ public:
     {
         sharedCtx->getSetResult(result, __isAllResult, __resultBytes, __result, elemType, elemSize);
     }
-
+    virtual IRowStream *getDatasetResult(IEngineRowAllocator * _resultAllocator)
+    {
+        UNIMPLEMENTED;
+    }
     virtual void bindBooleanParam(const char *name, bool val)
     {
         if (*argsig != 'B')

plugins/pyembed/CMakeLists.txt

@@ -39,6 +39,7 @@ if (USE_PYTHON)
          ./../../system/include
          ./../../rtl/eclrtl
          ./../../rtl/include
+         ./../../rtl/nbcd
          ./../../common/deftype
          ./../../system/jlib
        )

plugins/pyembed/pyembed.cpp

@@ -21,8 +21,12 @@
 #include "jthread.hpp"
 #include "hqlplugins.hpp"
 #include "deftype.hpp"
+#include "eclhelper.hpp"
 #include "eclrtl.hpp"
 #include "eclrtl_imp.hpp"
+#include "rtlds_imp.hpp"
+#include "rtlfield_imp.hpp"
+#include "nbcd.hpp"
 
 #ifdef _WIN32
 #define EXPORT __declspec(dllexport)
@@ -69,6 +73,7 @@ public:
     inline OwnedPyObject(PyObject *_ptr) : ptr(_ptr) {}
     inline ~OwnedPyObject()                { if (ptr) Py_DECREF(ptr); }
     inline PyObject * get() const           { return ptr; }
+    inline PyObject * getClear()            { PyObject *ret = ptr; ptr = NULL; return ret; }
     inline PyObject * operator -> () const { return ptr; }
     inline operator PyObject *() const    { return ptr; }
     inline void clear()                     { if (ptr) Py_DECREF(ptr); ptr = NULL; }
@@ -304,15 +309,421 @@ protected:
     HINSTANCE pythonLibrary;
 } globalState;
 
+static int countFields(const RtlFieldInfo * const * fields)
+{
+    unsigned count = 0;
+    loop
+    {
+        if (!*fields)
+            break;
+        fields++;
+        count++;
+    }
+    return count;
+}
+
+// Conversions from Python objects to ECL data
+
+static void typeError(const char *expected, const RtlFieldInfo *field) __attribute__((noreturn));
+
+static void typeError(const char *expected, const RtlFieldInfo *field)
+{
+    VStringBuffer msg("pyembed: type mismatch - %s expected", expected);
+    if (field)
+        msg.appendf(" for field %s", field->name->str());
+    rtlFail(0, msg.str());
+}
+
+static bool getBooleanResult(const RtlFieldInfo *field, PyObject *obj)
+{
+    assertex(obj && obj != Py_None);
+    if (!PyBool_Check(obj))
+        typeError("boolean", field);
+    return obj == Py_True;
+}
+
+static void getDataResult(const RtlFieldInfo *field, PyObject *obj, size32_t &chars, void * &result)
+{
+    assertex(obj && obj != Py_None);
+    if (!PyByteArray_Check(obj))
+        typeError("bytearray", field);
+    rtlStrToDataX(chars, result, PyByteArray_Size(obj), PyByteArray_AsString(obj));
+}
+
+static double getRealResult(const RtlFieldInfo *field, PyObject *obj)
+{
+    assertex(obj && obj != Py_None);
+    if (!PyFloat_Check(obj))
+        typeError("real", field);
+    return PyFloat_AsDouble(obj);
+}
+
+static __int64 getSignedResult(const RtlFieldInfo *field, PyObject *obj)
+{
+    assertex(obj && obj != Py_None);
+    __int64 ret;
+    if (PyInt_Check(obj))
+        ret = PyInt_AsUnsignedLongLongMask(obj);
+    else if (PyLong_Check(obj))
+        ret = (__int64) PyLong_AsLongLong(obj);
+    else
+        typeError("integer", field);
+    return ret;
+}
+
+static unsigned __int64 getUnsignedResult(const RtlFieldInfo *field, PyObject *obj)
+{
+    assertex(obj && obj != Py_None);
+    unsigned __int64 ret;
+    if (PyInt_Check(obj))
+        ret = PyInt_AsUnsignedLongLongMask(obj);
+    else if (PyLong_Check(obj))
+        ret =  (unsigned __int64) PyLong_AsUnsignedLongLong(obj);
+    else
+        typeError("integer", field);
+    return ret;
+}
+
+static void getStringResult(const RtlFieldInfo *field, PyObject *obj, size32_t &chars, char * &result)
+{
+    assertex(obj && obj != Py_None);
+    if (PyString_Check(obj))
+    {
+        const char * text =  PyString_AsString(obj);
+        checkPythonError();
+        size_t lenBytes = PyString_Size(obj);
+        rtlStrToStrX(chars, result, lenBytes, text);
+    }
+    else
+        typeError("string", field);
+}
+
+static void getUTF8Result(const RtlFieldInfo *field, PyObject *obj, size32_t &chars, char * &result)
+{
+    assertex(obj && obj != Py_None);
+    if (PyUnicode_Check(obj))
+    {
+        OwnedPyObject utf8 = PyUnicode_AsUTF8String(obj);
+        checkPythonError();
+        size_t lenBytes = PyString_Size(utf8);
+        const char * text =  PyString_AsString(utf8);
+        checkPythonError();
+        size32_t numchars = rtlUtf8Length(lenBytes, text);
+        rtlUtf8ToUtf8X(chars, result, numchars, text);
+    }
+    else
+        typeError("unicode string", field);
+}
+
+static void getSetResult(PyObject *obj, bool & isAllResult, size32_t & resultBytes, void * & result, int elemType, size32_t elemSize)
+{
+    // MORE - should probably recode to use the getResultDataset mechanism
+    assertex(obj && obj != Py_None);
+    if (!PyList_Check(obj) && !PySet_Check(obj))
+        rtlFail(0, "pyembed: type mismatch - list or set expected");
+    rtlRowBuilder out;
+    size32_t outBytes = 0;
+    byte *outData = NULL;
+    OwnedPyObject iter = PyObject_GetIter(obj);
+    OwnedPyObject elem;
+    for (elem.setown(PyIter_Next(iter)); elem != NULL; elem.setown(PyIter_Next(iter)))
+    {
+        if (elemSize != UNKNOWN_LENGTH)
+        {
+            out.ensureAvailable(outBytes + elemSize);
+            outData = out.getbytes() + outBytes;
+            outBytes += elemSize;
+        }
+        switch ((type_t) elemType)
+        {
+        case type_int:
+            rtlWriteInt(outData, pyembed::getSignedResult(NULL, elem), elemSize);
+            break;
+        case type_unsigned:
+            rtlWriteInt(outData, pyembed::getUnsignedResult(NULL, elem), elemSize);
+            break;
+        case type_real:
+            if (elemSize == sizeof(double))
+                * (double *) outData = (double) pyembed::getRealResult(NULL, elem);
+            else
+            {
+                assertex(elemSize == sizeof(float));
+                * (float *) outData = (float) pyembed::getRealResult(NULL, elem);
+            }
+            break;
+        case type_boolean:
+            assertex(elemSize == sizeof(bool));
+            * (bool *) outData = pyembed::getBooleanResult(NULL, elem);
+            break;
+        case type_string:
+        case type_varstring:
+        {
+            if (!PyString_Check(elem))
+                rtlFail(0, "pyembed: type mismatch - return value in list was not a STRING");
+            const char * text =  PyString_AsString(elem);
+            checkPythonError();
+            size_t lenBytes = PyString_Size(elem);
+            if (elemSize == UNKNOWN_LENGTH)
+            {
+                if (elemType == type_string)
+                {
+                    out.ensureAvailable(outBytes + lenBytes + sizeof(size32_t));
+                    outData = out.getbytes() + outBytes;
+                    * (size32_t *) outData = lenBytes;
+                    rtlStrToStr(lenBytes, outData+sizeof(size32_t), lenBytes, text);
+                    outBytes += lenBytes + sizeof(size32_t);
+                }
+                else
+                {
+                    out.ensureAvailable(outBytes + lenBytes + 1);
+                    outData = out.getbytes() + outBytes;
+                    rtlStrToVStr(0, outData, lenBytes, text);
+                    outBytes += lenBytes + 1;
+                }
+            }
+            else
+            {
+                if (elemType == type_string)
+                    rtlStrToStr(elemSize, outData, lenBytes, text);
+                else
+                    rtlStrToVStr(elemSize, outData, lenBytes, text);  // Fixed size null terminated strings... weird.
+            }
+            break;
+        }
+        case type_unicode:
+        case type_utf8:
+        {
+            if (!PyUnicode_Check(elem))
+                rtlFail(0, "pyembed: type mismatch - return value in list was not a unicode STRING");
+            OwnedPyObject utf8 = PyUnicode_AsUTF8String(elem);
+            checkPythonError();
+            size_t lenBytes = PyString_Size(utf8);
+            const char * text =  PyString_AsString(utf8);
+            checkPythonError();
+            size32_t numchars = rtlUtf8Length(lenBytes, text);
+            if (elemType == type_utf8)
+            {
+                assertex (elemSize == UNKNOWN_LENGTH);
+                out.ensureAvailable(outBytes + lenBytes + sizeof(size32_t));
+                outData = out.getbytes() + outBytes;
+                * (size32_t *) outData = numchars;
+                rtlStrToStr(lenBytes, outData+sizeof(size32_t), lenBytes, text);
+                outBytes += lenBytes + sizeof(size32_t);
+            }
+            else
+            {
+                if (elemSize == UNKNOWN_LENGTH)
+                {
+                    out.ensureAvailable(outBytes + numchars*sizeof(UChar) + sizeof(size32_t));
+                    outData = out.getbytes() + outBytes;
+                    // You can't assume that number of chars in utf8 matches number in unicode16 ...
+                    size32_t numchars16;
+                    rtlDataAttr unicode16;
+                    rtlUtf8ToUnicodeX(numchars16, unicode16.refustr(), numchars, text);
+                    * (size32_t *) outData = numchars16;
+                    rtlUnicodeToUnicode(numchars16, (UChar *) (outData+sizeof(size32_t)), numchars16, unicode16.getustr());
+                    outBytes += numchars16*sizeof(UChar) + sizeof(size32_t);
+                }
+                else
+                    rtlUtf8ToUnicode(elemSize / sizeof(UChar), (UChar *) outData, numchars, text);
+            }
+            break;
+        }
+        case type_data:
+        {
+            if (!PyByteArray_Check(elem))
+                rtlFail(0, "pyembed: type mismatch - return value in list was not a bytearray");
+            size_t lenBytes = PyByteArray_Size(elem);  // Could check does not overflow size32_t
+            const char *data = PyByteArray_AsString(elem);
+            if (elemSize == UNKNOWN_LENGTH)
+            {
+                out.ensureAvailable(outBytes + lenBytes + sizeof(size32_t));
+                outData = out.getbytes() + outBytes;
+                * (size32_t *) outData = lenBytes;
+                rtlStrToData(lenBytes, outData+sizeof(size32_t), lenBytes, data);
+                outBytes += lenBytes + sizeof(size32_t);
+            }
+            else
+                rtlStrToData(elemSize, outData, lenBytes, data);
+            break;
+        }
+        default:
+            rtlFail(0, "pyembed: type mismatch - unsupported return type");
+            break;
+        }
+        checkPythonError();
+    }
+    isAllResult = false;
+    resultBytes = outBytes;
+    result = out.detachdata();
+}
+
+static void getUnicodeResult(const RtlFieldInfo *field, PyObject *obj, size32_t &chars, UChar * &result)
+{
+    assertex(obj && obj != Py_None);
+    if (PyUnicode_Check(obj))
+    {
+        OwnedPyObject utf8 = PyUnicode_AsUTF8String(obj);
+        checkPythonError();
+        size_t lenBytes = PyString_Size(utf8);
+        const char * text =  PyString_AsString(utf8);
+        checkPythonError();
+        size32_t numchars = rtlUtf8Length(lenBytes, text);
+        rtlUtf8ToUnicodeX(chars, result, numchars, text);
+    }
+    else
+        typeError("unicode string", field);
+}
+
+// A PythonRowBuilder object is used to construct an ECL row from a python object
+
+class PythonRowBuilder : public CInterfaceOf<IFieldSource>
+{
+public:
+    PythonRowBuilder(PyObject *_row)
+    : iter(NULL), elem(_row)
+    {
+    }
+    virtual bool getBooleanResult(const RtlFieldInfo *field)
+    {
+        bool ret = pyembed::getBooleanResult(field, elem);
+        nextField();
+        return ret;
+    }
+    virtual void getDataResult(const RtlFieldInfo *field, size32_t &len, void * &result)
+    {
+        pyembed::getDataResult(field, elem, len, result);
+        nextField();
+    }
+    virtual double getRealResult(const RtlFieldInfo *field)
+    {
+        double ret = pyembed::getRealResult(field, elem);
+        nextField();
+        return ret;
+    }
+    virtual __int64 getSignedResult(const RtlFieldInfo *field)
+    {
+        __int64 ret = pyembed::getSignedResult(field, elem);
+        nextField();
+        return ret;
+    }
+    virtual unsigned __int64 getUnsignedResult(const RtlFieldInfo *field)
+    {
+        unsigned __int64 ret = pyembed::getUnsignedResult(field, elem);
+        nextField();
+        return ret;
+    }
+    virtual void getStringResult(const RtlFieldInfo *field, size32_t &chars, char * &result)
+    {
+        pyembed::getStringResult(field, elem, chars, result);
+        nextField();
+    }
+    virtual void getUTF8Result(const RtlFieldInfo *field, size32_t &chars, char * &result)
+    {
+        pyembed::getUTF8Result(field, elem, chars, result);
+        nextField();
+    }
+    virtual void getUnicodeResult(const RtlFieldInfo *field, size32_t &chars, UChar * &result)
+    {
+        pyembed::getUnicodeResult(field, elem, chars, result);
+        nextField();
+    }
+    virtual void getDecimalResult(const RtlFieldInfo *field, Decimal &value)
+    {
+        double ret = pyembed::getRealResult(field, elem);
+        value.setReal(ret);
+        nextField();
+    }
+
+    virtual void processBeginSet(const RtlFieldInfo * field, bool &isAll)
+    {
+        isAll = false;  // No concept of an 'all' set in Python
+        assertex(elem && elem != Py_None);
+        if (!PyList_Check(elem) && !PySet_Check(elem))
+            typeError("list or set", field);
+        iterStack.append(iter.getClear());
+        iter.setown(PyObject_GetIter(elem));
+        nextField();
+    }
+    virtual bool processNextSet(const RtlFieldInfo * field)
+    {
+        return elem != NULL;
+    }
+    virtual void processBeginDataset(const RtlFieldInfo * field)
+    {
+        if (PyList_Check(elem))
+        {
+            iterStack.append(iter.getClear());
+            iter.setown(PyObject_GetIter(elem));
+            nextField();
+        }
+        else
+            typeError("list", field);
+    }
+    virtual void processBeginRow(const RtlFieldInfo * field)
+    {
+        // Expect to see a tuple here, or possibly (if the ECL record has a single field), an arbitrary scalar object
+        // If it's a tuple, we push it onto our stack as the active object
+        iterStack.append(iter.getClear());
+        if (PyTuple_Check(elem))
+        {
+            iter.setown(PyObject_GetIter(elem));
+            nextField();
+        }
+        else if (countFields(field->type->queryFields())==1)
+        {
+            // iter is NULL;
+            // NOTE - we don't call nextField here. There is a single field. Debatable whether supporting this is helpful?
+        }
+        else
+        {
+            typeError("tuple", field);
+        }
+    }
+    virtual bool processNextRow(const RtlFieldInfo * field)
+    {
+        return elem != NULL;
+    }
+    virtual void processEndSet(const RtlFieldInfo * field)
+    {
+        iter.setown((PyObject *) iterStack.pop());
+        nextField();
+    }
+    virtual void processEndDataset(const RtlFieldInfo * field)
+    {
+        iter.setown((PyObject *) iterStack.pop());
+        nextField();
+    }
+    virtual void processEndRow(const RtlFieldInfo * field)
+    {
+        iter.setown((PyObject *) iterStack.pop());
+        nextField();
+    }
+protected:
+    void nextField()
+    {
+        if (iter)
+            elem.setown(PyIter_Next(iter));
+        else
+            elem = NULL;
+        checkPythonError();
+    }
+    OwnedPyObject iter;
+    OwnedPyObject elem;
+    PointerArray iterStack;
+};
+
 // Each call to a Python function will use a new Python27EmbedFunctionContext object
 // This takes care of ensuring that the Python GIL is locked while we are executing python code,
 // and released when we are not
 
-class Python27EmbedContextBase : public CInterfaceOf<IEmbedFunctionContext>
+class Python27EmbedContextBase : public CInterface, implements IEmbedFunctionContext, implements IRowStream
 {
 public:
+    IMPLEMENT_IINTERFACE;
+
     Python27EmbedContextBase(PythonThreadContext *_sharedCtx)
-    : sharedCtx(_sharedCtx)
+    : sharedCtx(_sharedCtx), resultIterator(NULL)
     {
         PyEval_RestoreThread(sharedCtx->threadState);
         locals.setown(PyDict_New());
@@ -331,225 +742,67 @@ public:
 
     virtual bool getBooleanResult()
     {
-        assertex(result && result != Py_None);
-        if (!PyBool_Check(result))
-            rtlFail(0, "pyembed: Type mismatch on result - value is not BOOLEAN ");
-        return result == Py_True;
+        return pyembed::getBooleanResult(NULL, result);
     }
     virtual void getDataResult(size32_t &__chars, void * &__result)
     {
-        assertex(result && result != Py_None);
-        if (!PyByteArray_Check(result))
-            rtlFail(0, "pyembed: Type mismatch on result - value is not a bytearray");
-        rtlStrToDataX(__chars, __result, PyByteArray_Size(result), PyByteArray_AsString(result));
+        pyembed::getDataResult(NULL, result, __chars, __result);
     }
     virtual double getRealResult()
     {
-        assertex(result && result != Py_None);
-        return PyFloat_AsDouble(result);
+        return pyembed::getRealResult(NULL, result);
     }
     virtual __int64 getSignedResult()
     {
-        assertex(result && result != Py_None);
-        return (__int64) PyLong_AsLongLong(result);
+        return pyembed::getSignedResult(NULL, result);
     }
     virtual unsigned __int64 getUnsignedResult()
     {
-        assertex(result && result != Py_None);
-        return (__int64) PyLong_AsUnsignedLongLong(result);
+        return pyembed::getUnsignedResult(NULL, result);
     }
     virtual void getStringResult(size32_t &__chars, char * &__result)
     {
-        assertex(result && result != Py_None);
-        if (PyString_Check(result))
-        {
-            const char * text =  PyString_AsString(result);
-            checkPythonError();
-            size_t lenBytes = PyString_Size(result);
-            rtlStrToStrX(__chars, __result, lenBytes, text);
-        }
-        else
-            rtlFail(0, "pyembed: type mismatch - return value was not a string");
+        pyembed::getStringResult(NULL, result, __chars, __result);
     }
     virtual void getUTF8Result(size32_t &__chars, char * &__result)
     {
-        assertex(result && result != Py_None);
-        if (PyUnicode_Check(result))
-        {
-            OwnedPyObject utf8 = PyUnicode_AsUTF8String(result);
-            checkPythonError();
-            size_t lenBytes = PyString_Size(utf8);
-            const char * text =  PyString_AsString(utf8);
-            checkPythonError();
-            size32_t numchars = rtlUtf8Length(lenBytes, text);
-            rtlUtf8ToUtf8X(__chars, __result, numchars, text);
-        }
-        else
-            rtlFail(0, "pyembed: type mismatch - return value was not a unicode string");
+        pyembed::getUTF8Result(NULL, result, __chars, __result);
     }
     virtual void getUnicodeResult(size32_t &__chars, UChar * &__result)
     {
-        assertex(result && result != Py_None);
-        if (PyUnicode_Check(result))
-        {
-            OwnedPyObject utf8 = PyUnicode_AsUTF8String(result);
-            checkPythonError();
-            size_t lenBytes = PyString_Size(utf8);
-            const char * text =  PyString_AsString(utf8);
-            checkPythonError();
-            size32_t numchars = rtlUtf8Length(lenBytes, text);
-            rtlUtf8ToUnicodeX(__chars, __result, numchars, text);
-        }
-        else
-            rtlFail(0, "pyembed: type mismatch - return value was not a unicode string");
+        pyembed::getUnicodeResult(NULL, result, __chars, __result);
     }
     virtual void getSetResult(bool & __isAllResult, size32_t & __resultBytes, void * & __result, int elemType, size32_t elemSize)
     {
+        pyembed::getSetResult(result, __isAllResult, __resultBytes, __result, elemType, elemSize);
+    }
+    virtual IRowStream *getDatasetResult(IEngineRowAllocator * _resultAllocator)
+    {
         assertex(result && result != Py_None);
         if (!PyList_Check(result))
-            rtlFail(0, "pyembed: type mismatch - return value was not a list");
-        Py_ssize_t numResults = PyList_Size(result);
-        rtlRowBuilder out;
-        byte *outData = NULL;
-        size32_t outBytes = 0;
-        if (elemSize != UNKNOWN_LENGTH)
-        {
-            out.ensureAvailable(numResults * elemSize); // MORE - check for overflow?
-            outData = out.getbytes();
-        }
-        for (int i = 0; i < numResults; i++)
-        {
-            PyObject *elem = PyList_GetItem(result, i); // note - borrowed reference
-            switch ((type_t) elemType)
-            {
-            case type_int:
-                rtlWriteInt(outData, PyLong_AsLongLong(elem), elemSize);
-                break;
-            case type_unsigned:
-                rtlWriteInt(outData, PyLong_AsUnsignedLongLong(elem), elemSize);
-                break;
-            case type_real:
-                if (!PyFloat_Check(elem))
-                    rtlFail(0, "pyembed: type mismatch - return value in list was not a REAL");
-                if (elemSize == sizeof(double))
-                    * (double *) outData = (double) PyFloat_AsDouble(elem);
-                else
-                {
-                    assertex(elemSize == sizeof(float));
-                    * (float *) outData = (float) PyFloat_AsDouble(elem);
-                }
-                break;
-            case type_boolean:
-                assertex(elemSize == sizeof(bool));
-                if (!PyBool_Check(elem))
-                    rtlFail(0, "pyembed: type mismatch - return value in list was not a BOOLEAN");
-                * (bool *) outData = (result == Py_True);
-                break;
-            case type_string:
-            case type_varstring:
-            {
-                if (!PyString_Check(elem))
-                    rtlFail(0, "pyembed: type mismatch - return value in list was not a STRING");
-                const char * text =  PyString_AsString(elem);
-                checkPythonError();
-                size_t lenBytes = PyString_Size(elem);
-                if (elemSize == UNKNOWN_LENGTH)
-                {
-                    if (elemType == type_string)
-                    {
-                        out.ensureAvailable(outBytes + lenBytes + sizeof(size32_t));
-                        outData = out.getbytes() + outBytes;
-                        * (size32_t *) outData = lenBytes;
-                        rtlStrToStr(lenBytes, outData+sizeof(size32_t), lenBytes, text);
-                        outBytes += lenBytes + sizeof(size32_t);
-                    }
-                    else
-                    {
-                        out.ensureAvailable(outBytes + lenBytes + 1);
-                        outData = out.getbytes() + outBytes;
-                        rtlStrToVStr(0, outData, lenBytes, text);
-                        outBytes += lenBytes + 1;
-                    }
-                }
-                else
-                {
-                    if (elemType == type_string)
-                        rtlStrToStr(elemSize, outData, lenBytes, text);
-                    else
-                        rtlStrToVStr(elemSize, outData, lenBytes, text);  // Fixed size null terminated strings... weird.
-                }
-                break;
-            }
-            case type_unicode:
-            case type_utf8:
-            {
-                if (!PyUnicode_Check(elem))
-                    rtlFail(0, "pyembed: type mismatch - return value in list was not a unicode STRING");
-                OwnedPyObject utf8 = PyUnicode_AsUTF8String(elem);
-                checkPythonError();
-                size_t lenBytes = PyString_Size(utf8);
-                const char * text =  PyString_AsString(utf8);
-                checkPythonError();
-                size32_t numchars = rtlUtf8Length(lenBytes, text);
-                if (elemType == type_utf8)
-                {
-                    assertex (elemSize == UNKNOWN_LENGTH);
-                    out.ensureAvailable(outBytes + lenBytes + sizeof(size32_t));
-                    outData = out.getbytes() + outBytes;
-                    * (size32_t *) outData = numchars;
-                    rtlStrToStr(lenBytes, outData+sizeof(size32_t), lenBytes, text);
-                    outBytes += lenBytes + sizeof(size32_t);
-                }
-                else
-                {
-                    if (elemSize == UNKNOWN_LENGTH)
-                    {
-                        out.ensureAvailable(outBytes + numchars*sizeof(UChar) + sizeof(size32_t));
-                        outData = out.getbytes() + outBytes;
-                        // You can't assume that number of chars in utf8 matches number in unicode16 ...
-                        size32_t numchars16;
-                        rtlDataAttr unicode16;
-                        rtlUtf8ToUnicodeX(numchars16, unicode16.refustr(), numchars, text);
-                        * (size32_t *) outData = numchars16;
-                        rtlUnicodeToUnicode(numchars16, (UChar *) (outData+sizeof(size32_t)), numchars16, unicode16.getustr());
-                        outBytes += numchars16*sizeof(UChar) + sizeof(size32_t);
-                    }
-                    else
-                        rtlUtf8ToUnicode(elemSize / sizeof(UChar), (UChar *) outData, numchars, text);
-                }
-                break;
-            }
-            case type_data:
-            {
-                if (!PyByteArray_Check(elem))
-                    rtlFail(0, "pyembed: type mismatch - return value in list was not a bytearray");
-                size_t lenBytes = PyByteArray_Size(elem);  // Could check does not overflow size32_t
-                const char *data = PyByteArray_AsString(elem);
-                if (elemSize == UNKNOWN_LENGTH)
-                {
-                    out.ensureAvailable(outBytes + lenBytes + sizeof(size32_t));
-                    outData = out.getbytes() + outBytes;
-                    * (size32_t *) outData = lenBytes;
-                    rtlStrToData(lenBytes, outData+sizeof(size32_t), lenBytes, data);
-                    outBytes += lenBytes + sizeof(size32_t);
-                }
-                else
-                    rtlStrToData(elemSize, outData, lenBytes, data);
-                break;
-            }
-            default:
-                rtlFail(0, "pyembed: type mismatch - unsupported return type");
-            }
-            checkPythonError();
-            if (elemSize != UNKNOWN_LENGTH)
-            {
-                outData += elemSize;
-                outBytes += elemSize;
-            }
-        }
-        __isAllResult = false;
-        __resultBytes = outBytes;
-        __result = out.detachdata();
+            typeError("list", NULL);
+        resultIterator = PyObject_GetIter(result);
+        resultAllocator.set(_resultAllocator);
+        return LINK(this);
+    }
+    virtual const void *nextRow()
+    {
+        assertex(resultAllocator);
+        assertex(resultIterator);
+        OwnedPyObject row = PyIter_Next(resultIterator);
+        if (!row)
+            return NULL;
+        RtlDynamicRowBuilder rowBuilder(resultAllocator);
+        PythonRowBuilder pyRowBuilder(row);
+        const RtlTypeInfo *typeInfo = resultAllocator->queryOutputMeta()->queryTypeInfo();
+        assertex(typeInfo);
+        RtlFieldStrInfo dummyField("<row>", NULL, typeInfo);
+        size32_t len = typeInfo->build(rowBuilder, 0, &dummyField, pyRowBuilder);
+        return rowBuilder.finalizeRowClear(len);
+    }
+    virtual void stop()
+    {
+        resultAllocator.clear();
     }
 
     virtual void bindBooleanParam(const char *name, bool val)
@@ -694,6 +947,9 @@ protected:
     OwnedPyObject globals;
     OwnedPyObject result;
     OwnedPyObject script;
+
+    Linked<IEngineRowAllocator> resultAllocator;
+    PyObject *resultIterator;
 };
 
 class Python27EmbedScriptContext : public Python27EmbedContextBase

plugins/v8embed/v8embed.cpp

@@ -435,6 +435,11 @@ public:
         __result = out.detachdata();
     }
 
+    virtual IRowStream *getDatasetResult(IEngineRowAllocator * _resultAllocator)
+    {
+        UNIMPLEMENTED;
+    }
+
     virtual void compileEmbeddedScript(size32_t lenChars, const char *utf)
     {
         v8::HandleScope handle_scope;

rtl/eclrtl/eclrtl.hpp

@@ -450,10 +450,15 @@ ECLRTL_API unsigned __int64 rtlReadSwapUInt7(const void * data);
 ECLRTL_API unsigned __int64 rtlReadSwapUInt8(const void * data);
 ECLRTL_API unsigned __int64 rtlReadSwapUInt(const void * data, unsigned length);
 
+inline void rtlWriteSwapInt1(void * data, unsigned value) { *(unsigned char *)data = value; }
+ECLRTL_API void rtlWriteSwapInt2(void * data, unsigned value);
 ECLRTL_API void rtlWriteSwapInt3(void * data, unsigned value);
+ECLRTL_API void rtlWriteSwapInt4(void * data, unsigned value);
 ECLRTL_API void rtlWriteSwapInt5(void * data, unsigned __int64 value);
 ECLRTL_API void rtlWriteSwapInt6(void * data, unsigned __int64 value);
 ECLRTL_API void rtlWriteSwapInt7(void * data, unsigned __int64 value);
+ECLRTL_API void rtlWriteSwapInt8(void * data, unsigned __int64 value);
+ECLRTL_API void rtlWriteSwapInt(void * self, __int64 val, unsigned length);
 
 ECLRTL_API short rtlRevInt2(const void * data);
 ECLRTL_API int rtlRevInt3(const void * data);
@@ -762,6 +767,8 @@ ECLRTL_API unsigned rtlDelayReturn(unsigned value, unsigned sleepTime);
 ECLRTL_API bool rtlGPF();
 
 //-----------------------------------------------------------------------------
+interface IRowStream;
+struct RtlTypeInfo;
 
 interface IEmbedFunctionContext : extends IInterface
 {
@@ -790,6 +797,9 @@ interface IEmbedFunctionContext : extends IInterface
     virtual void importFunction(size32_t len, const char *function) = 0;
     virtual void compileEmbeddedScript(size32_t len, const char *script) = 0;
     virtual void callFunction() = 0;
+
+    virtual IRowStream *getDatasetResult(IEngineRowAllocator * _resultAllocator) = 0;
+
 };
 
 interface IEmbedContext : extends IInterface

rtl/eclrtl/rtlfield.cpp

@@ -23,6 +23,8 @@
 #include "eclhelper.hpp"
 #include "eclrtl_imp.hpp"
 #include "rtlfield_imp.hpp"
+#include "rtlds_imp.hpp"
+#include "nbcd.hpp"
 
 static const char * queryXPath(const RtlFieldInfo * field)
 {
@@ -73,7 +75,13 @@ size32_t RtlTypeInfoBase::toXML(const byte * self, const byte * selfrow, const R
     rtlFailUnexpected();
     return 0;
 }
-
+/*
+size32_t RtlTypeInfoBase::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    rtlFailUnexpected();
+    return 0;
+}
+*/
 const char * RtlTypeInfoBase::queryLocale() const 
 {
     return NULL; 
@@ -91,6 +99,15 @@ const RtlTypeInfo * RtlTypeInfoBase::queryChildType() const
 
 //-------------------------------------------------------------------------------------------------------------------
 
+size32_t RtlBoolTypeInfo::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    builder.ensureCapacity(sizeof(bool)+offset, field->name->str());
+    bool val = source.getBooleanResult(field);
+    * (bool *) (builder.getSelf() + offset) = val;
+    offset += sizeof(bool);
+    return offset;
+}
+
 size32_t RtlBoolTypeInfo::process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const
 {
     target.processBool(*(const bool *)self, field);
@@ -112,6 +129,19 @@ double RtlRealTypeInfo::value(const byte * self) const
     return *(const double *)self;
 }
 
+size32_t RtlRealTypeInfo::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    builder.ensureCapacity(length+offset, field->name->str());
+    double val = source.getRealResult(field);
+    byte *dest = builder.getSelf() + offset;
+    if (length == 4)
+        *(float *) dest = (float) val;
+    else
+        *(double *) dest = val;
+    offset += length;
+    return offset;
+}
+
 size32_t RtlRealTypeInfo::process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const
 {
     target.processReal(value(self), field);
@@ -126,6 +156,15 @@ size32_t RtlRealTypeInfo::toXML(const byte * self, const byte * selfrow, const R
 
 //-------------------------------------------------------------------------------------------------------------------
 
+size32_t RtlIntTypeInfo::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    builder.ensureCapacity(length+offset, field->name->str());
+    __int64 val = isUnsigned() ? (__int64) source.getUnsignedResult(field) : source.getSignedResult(field);
+    rtlWriteInt(builder.getSelf() + offset, val, length);
+    offset += length;
+    return offset;
+}
+
 size32_t RtlIntTypeInfo::process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const
 {
     if (isUnsigned())
@@ -146,6 +185,15 @@ size32_t RtlIntTypeInfo::toXML(const byte * self, const byte * selfrow, const Rt
 
 //-------------------------------------------------------------------------------------------------------------------
 
+size32_t RtlSwapIntTypeInfo::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    builder.ensureCapacity(length+offset, field->name->str());
+    __int64 val = isUnsigned() ? (__int64) source.getUnsignedResult(field) : source.getSignedResult(field);
+    rtlWriteSwapInt(builder.getSelf() + offset, val, length);
+    offset += length;
+    return offset;
+}
+
 size32_t RtlSwapIntTypeInfo::process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const
 {
     if (isUnsigned())
@@ -171,6 +219,16 @@ size32_t RtlPackedIntTypeInfo::size(const byte * self, const byte * selfrow) con
     return rtlGetPackedSize(self); 
 }
 
+size32_t RtlPackedIntTypeInfo::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    unsigned __int64 value = isUnsigned() ? (__int64) source.getUnsignedResult(field) : source.getSignedResult(field);
+    size32_t sizeInBytes = rtlGetPackedSize(&value);
+    builder.ensureCapacity(sizeInBytes+offset, field->name->str());
+    rtlSetPackedUnsigned(builder.getSelf() + offset, value);
+    offset += sizeInBytes;
+    return offset;
+}
+
 size32_t RtlPackedIntTypeInfo::process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const
 {
     if (isUnsigned())
@@ -198,6 +256,36 @@ size32_t RtlStringTypeInfo::size(const byte * self, const byte * selfrow) const
     return sizeof(size32_t) + rtlReadUInt4(self);
 }
 
+size32_t RtlStringTypeInfo::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    size32_t size;
+    char *value;
+    source.getStringResult(field, size, value);
+    if (!isFixedSize())
+    {
+        builder.ensureCapacity(offset+size+sizeof(size32_t), field->name->str());
+        byte *dest = builder.getSelf()+offset;
+        rtlWriteInt4(dest, size);
+        if (isEbcdic())
+            rtlStrToEStr(size, (char *) dest+sizeof(size32_t), size, (char *)value);  // slightly debatable - might expect incoming result to already be in ebcdic?
+        else
+            memcpy(dest+sizeof(size32_t), value, size);
+        offset += size+sizeof(size32_t);
+    }
+    else
+    {
+        builder.ensureCapacity(offset+length, field->name->str());
+        byte *dest = builder.getSelf()+offset;
+        if (isEbcdic())
+            rtlStrToEStr(length, (char *) dest, size, (char *) value);
+        else
+            rtlStrToStr(length, dest, size, value);
+        offset += length;
+    }
+    rtlFree(value);
+    return offset;
+}
+
 size32_t RtlStringTypeInfo::process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const
 {
     const char * str = reinterpret_cast<const char *>(self);
@@ -269,6 +357,30 @@ size32_t RtlDataTypeInfo::size(const byte * self, const byte * selfrow) const
     return sizeof(size32_t) + rtlReadUInt4(self);
 }
 
+size32_t RtlDataTypeInfo::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    size32_t size;
+    void *value;
+    source.getDataResult(field, size, value);
+    if (!isFixedSize())
+    {
+        builder.ensureCapacity(offset+size+sizeof(size32_t), field->name->str());
+        byte *dest = builder.getSelf()+offset;
+        rtlWriteInt4(dest, size);
+        memcpy(dest+sizeof(size32_t), value, size);
+        offset += size+sizeof(size32_t);
+    }
+    else
+    {
+        builder.ensureCapacity(offset+length, field->name->str());
+        byte *dest = builder.getSelf()+offset;
+        rtlDataToData(length, dest, size, value);
+        offset += length;
+    }
+    rtlFree(value);
+    return offset;
+}
+
 size32_t RtlDataTypeInfo::process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const
 {
     const char * str = reinterpret_cast<const char *>(self);
@@ -321,6 +433,31 @@ size32_t RtlVarStringTypeInfo::size(const byte * self, const byte * selfrow) con
     return (size32_t)strlen(str)+1;
 }
 
+size32_t RtlVarStringTypeInfo::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    size32_t size;
+    char *value;
+    source.getStringResult(field, size, value);
+    byte *dest = builder.getSelf()+offset;
+    if (isEbcdic())
+        UNIMPLEMENTED;
+    if (!isFixedSize())
+    {
+        builder.ensureCapacity(offset+size+1, field->name->str());
+        memcpy(dest, value, size);
+        dest[size] = '\0';
+        offset += size+1;
+    }
+    else
+    {
+        builder.ensureCapacity(offset+length, field->name->str());
+        rtlStrToVStr(length, dest, size, value);
+        offset += length;
+    }
+    rtlFree(value);
+    return offset;
+}
+
 size32_t RtlVarStringTypeInfo::process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const
 {
     const char * str = reinterpret_cast<const char *>(self);
@@ -376,6 +513,33 @@ size32_t RtlQStringTypeInfo::size(const byte * self, const byte * selfrow) const
     return sizeof(size32_t) + rtlQStrSize(rtlReadUInt4(self));
 }
 
+size32_t RtlQStringTypeInfo::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    size32_t size;
+    char *value;
+    source.getStringResult(field, size, value);
+    byte *dest = builder.getSelf()+offset;
+    if (!isFixedSize())
+    {
+        size32_t sizeInBytes = rtlQStrSize(size) + sizeof(size32_t);
+        builder.ensureCapacity(offset+sizeInBytes, field->name->str());
+        byte *dest = builder.getSelf()+offset;
+        rtlWriteInt4(dest, size);
+        rtlStrToQStr(size, (char *) dest+sizeof(size32_t), size, value);
+        offset += sizeInBytes;
+    }
+    else
+    {
+        size32_t sizeInBytes = rtlQStrSize(length);
+        builder.ensureCapacity(offset+sizeInBytes, field->name->str());
+        byte *dest = builder.getSelf()+offset;
+        rtlStrToQStr(length, (char *) dest, size, value);
+        offset += sizeInBytes;
+    }
+    rtlFree(value);
+    return offset;
+}
+
 size32_t RtlQStringTypeInfo::process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const
 {
     const char * str = reinterpret_cast<const char *>(self);
@@ -432,6 +596,20 @@ size32_t RtlDecimalTypeInfo::size(const byte * self, const byte * selfrow) const
     return calcSize();
 }
 
+size32_t RtlDecimalTypeInfo::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    Decimal value;
+    source.getDecimalResult(field, value);
+    size32_t sizeInBytes = calcSize();
+    builder.ensureCapacity(sizeInBytes+offset, field->name->str());
+    if (isUnsigned())
+        value.getUDecimal(sizeInBytes, getDecimalPrecision(), builder.getSelf()+offset);
+    else
+        value.getDecimal(sizeInBytes, getDecimalPrecision(), builder.getSelf()+offset);
+    offset += sizeInBytes;
+    return offset;
+}
+
 size32_t RtlDecimalTypeInfo::process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const
 {
     size32_t thisSize = calcSize();
@@ -454,6 +632,11 @@ size32_t RtlDecimalTypeInfo::toXML(const byte * self, const byte * selfrow, cons
 
 //-------------------------------------------------------------------------------------------------------------------
 
+size32_t RtlCharTypeInfo::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    throwUnexpected();  // Can't have a field of type char
+}
+
 size32_t RtlCharTypeInfo::process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const
 {
     const char * str = reinterpret_cast<const char *>(self);
@@ -487,6 +670,32 @@ size32_t RtlUnicodeTypeInfo::size(const byte * self, const byte * selfrow) const
     return sizeof(size32_t) + rtlReadUInt4(self) * sizeof(UChar);
 }
 
+size32_t RtlUnicodeTypeInfo::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    size32_t sizeInChars;
+    UChar *value;
+    source.getUnicodeResult(field, sizeInChars, value);
+    if (!isFixedSize())
+    {
+        size32_t sizeInBytes = sizeInChars * sizeof(UChar);
+        builder.ensureCapacity(offset+sizeInBytes+sizeof(size32_t), field->name->str());
+        byte *dest = builder.getSelf()+offset;
+        rtlWriteInt4(dest, sizeInChars);  // NOTE - in chars!
+        memcpy(dest+sizeof(size32_t), value, sizeInBytes);
+        offset += sizeInBytes+sizeof(size32_t);
+    }
+    else
+    {
+        size32_t sizeInBytes = length * sizeof(UChar);
+        builder.ensureCapacity(offset+sizeInBytes, field->name->str());
+        byte *dest = builder.getSelf()+offset;
+        rtlUnicodeToUnicode(length, (UChar *) dest, sizeInChars, value);
+        offset += sizeInBytes;
+    }
+    rtlFree(value);
+    return offset;
+}
+
 size32_t RtlUnicodeTypeInfo::process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const
 {
     const UChar * ustr = reinterpret_cast<const UChar *>(self);
@@ -539,6 +748,32 @@ size32_t RtlVarUnicodeTypeInfo::size(const byte * self, const byte * selfrow) co
     return (rtlUnicodeStrlen(ustr)+1) * sizeof(UChar);
 }
 
+size32_t RtlVarUnicodeTypeInfo::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    size32_t sizeInChars;
+    UChar *value;
+    source.getUnicodeResult(field, sizeInChars, value);
+    if (!isFixedSize())
+    {
+        size32_t sizeInBytes = (sizeInChars+1) * sizeof(UChar);
+        builder.ensureCapacity(offset+sizeInBytes, field->name->str());
+        UChar *dest = (UChar *) builder.getSelf()+offset;
+        memcpy(dest, value, sizeInBytes - sizeof(UChar));
+        dest[sizeInChars] = 0;
+        offset += sizeInBytes;
+    }
+    else
+    {
+        size32_t sizeInBytes = length * sizeof(UChar);
+        builder.ensureCapacity(offset+sizeInBytes, field->name->str());
+        byte *dest = builder.getSelf()+offset;
+        rtlUnicodeToVUnicode(length, (UChar *) dest, sizeInChars, value);
+        offset += sizeInBytes;
+    }
+    rtlFree(value);
+    return offset;
+}
+
 size32_t RtlVarUnicodeTypeInfo::process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const
 {
     const UChar * ustr = reinterpret_cast<const UChar *>(self);
@@ -575,6 +810,22 @@ size32_t RtlUtf8TypeInfo::size(const byte * self, const byte * selfrow) const
     return sizeof(size32_t) + rtlUtf8Size(rtlReadUInt4(self), self+sizeof(unsigned));
 }
 
+size32_t RtlUtf8TypeInfo::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    size32_t sizeInChars;
+    char *value;
+    source.getUTF8Result(field, sizeInChars, value);
+    size32_t sizeInBytes = rtlUtf8Size(sizeInChars, value);
+    assertex(!isFixedSize());
+    builder.ensureCapacity(offset+sizeInBytes+sizeof(size32_t), field->name->str());
+    byte *dest = builder.getSelf()+offset;
+    rtlWriteInt4(dest, sizeInChars);  // NOTE - in chars!
+    memcpy(dest+sizeof(size32_t), value, sizeInBytes);
+    offset += sizeInBytes+sizeof(size32_t);
+    rtlFree(value);
+    return offset;
+}
+
 size32_t RtlUtf8TypeInfo::process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const
 {
     assertex(!isFixedSize());
@@ -628,6 +879,20 @@ inline size32_t processFields(const RtlFieldInfo * const * cur, const byte * sel
     return offset;
 }
 
+inline size32_t processFields(const RtlFieldInfo * const * cur, ARowBuilder &builder, size32_t offset, IFieldSource &source)
+{
+    loop
+    {
+        const RtlFieldInfo * child = *cur;
+        if (!child)
+            break;
+        offset = child->build(builder, offset, source);
+        cur++;
+    }
+    return offset;
+}
+
+
 inline size32_t toXMLFields(const RtlFieldInfo * const * cur, const byte * self, const byte * selfrow, IXmlWriter & target)
 {
     size32_t offset = 0;
@@ -676,6 +941,14 @@ size32_t RtlRecordTypeInfo::toXML(const byte * self, const byte * selfrow, const
     return thisSize;
 }
 
+size32_t RtlRecordTypeInfo::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    source.processBeginRow(field);
+    offset = processFields(fields, builder, offset, source);
+    source.processEndRow(field);
+    return offset;
+}
+
 //-------------------------------------------------------------------------------------------------------------------
 
 size32_t RtlSetTypeInfo::size(const byte * self, const byte * selfrow) const 
@@ -683,6 +956,36 @@ size32_t RtlSetTypeInfo::size(const byte * self, const byte * selfrow) const
     return sizeof(bool) + sizeof(size32_t) + rtlReadUInt4(self + sizeof(bool));
 }
 
+size32_t RtlSetTypeInfo::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    bool isAll;
+    source.processBeginSet(field, isAll);
+    size32_t sizeInBytes = sizeof(bool) + sizeof(size32_t);
+    builder.ensureCapacity(offset+sizeInBytes, field->name->str());
+    byte *dest = builder.getSelf()+offset;
+    if (isAll)
+    {
+        * (bool *) dest = true;
+        rtlWriteInt4(dest+1, 0);
+        offset += sizeInBytes;
+    }
+    else
+    {
+        * (bool *) dest = false;
+        size32_t newOffset = offset + sizeInBytes;
+        RtlFieldStrInfo dummyField("<set element>", NULL, child);
+        while (source.processNextSet(field))
+        {
+            newOffset = child->build(builder, newOffset, &dummyField, source);
+        }
+        // Go back in and patch the size, remembering it may have moved
+        rtlWriteInt4(builder.getSelf()+offset+1, newOffset - (offset+sizeInBytes));
+        offset = newOffset;
+    }
+    source.processEndSet(field);
+    return offset;
+}
+
 size32_t RtlSetTypeInfo::process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const
 {
     unsigned offset = sizeof(bool) + sizeof(size32_t);
@@ -777,6 +1080,42 @@ size32_t RtlDatasetTypeInfo::size(const byte * self, const byte * selfrow) const
     return sizeof(size32_t) + rtlReadUInt4(self);
 }
 
+size32_t RtlDatasetTypeInfo::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    source.processBeginDataset(field);
+    if (isLinkCounted())
+    {
+        // a 32-bit record count, and a pointer to an array of record pointers
+        size32_t sizeInBytes = sizeof(size32_t) + sizeof(void *);
+        builder.ensureCapacity(offset+sizeInBytes, field->name->str());
+        size32_t newOffset = offset + sizeInBytes;
+        size32_t numRows = 0;
+
+        // MORE - read all the child rows
+
+        // Go back in and patch the count, remembering it may have moved
+        rtlWriteInt4(builder.getSelf()+offset, numRows);
+        // * ( void * * ) (builder.getSelf()+offset+sizeof(size32_t)) = rows;
+        offset = newOffset;
+    }
+    else
+    {
+        // a 32-bit size, then rows inline
+        size32_t sizeInBytes = sizeof(size32_t);
+        builder.ensureCapacity(offset+sizeInBytes, field->name->str());
+        size32_t newOffset = offset + sizeInBytes;
+        RtlFieldStrInfo dummyField("<nested row>", NULL, child);
+        while (source.processNextRow(field))
+            newOffset = child->build(builder, newOffset, &dummyField, source);
+        // Go back in and patch the size, remembering it may have moved
+        rtlWriteInt4(builder.getSelf()+offset, newOffset - (offset+sizeInBytes));
+        offset = newOffset;
+    }
+    source.processEndDataset(field);
+    return offset;
+
+}
+
 size32_t RtlDatasetTypeInfo::process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const
 {
     if (isLinkCounted())
@@ -863,6 +1202,16 @@ size32_t RtlDictionaryTypeInfo::size(const byte * self, const byte * selfrow) co
     return sizeof(size32_t) + rtlReadUInt4(self);
 }
 
+size32_t RtlDictionaryTypeInfo::build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const
+{
+    if (isLinkCounted())
+    {
+        UNIMPLEMENTED;
+    }
+    else
+        UNIMPLEMENTED;
+}
+
 size32_t RtlDictionaryTypeInfo::process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const
 {
     if (isLinkCounted())

rtl/eclrtl/rtlfield_imp.hpp

@@ -28,6 +28,8 @@ struct ECLRTL_API RtlTypeInfoBase : public RtlTypeInfo
     virtual size32_t size(const byte * self, const byte * selfrow) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
+    // virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
+
 
     virtual const char * queryLocale() const;
     virtual const RtlFieldInfo * const * queryFields() const;
@@ -40,6 +42,7 @@ struct ECLRTL_API RtlBoolTypeInfo : public RtlTypeInfoBase
 {
     inline RtlBoolTypeInfo(unsigned _fieldType, unsigned _length) : RtlTypeInfoBase(_fieldType, _length) {}
 
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
 };
@@ -48,6 +51,7 @@ struct ECLRTL_API RtlRealTypeInfo : public RtlTypeInfoBase
 {
     inline RtlRealTypeInfo(unsigned _fieldType, unsigned _length) : RtlTypeInfoBase(_fieldType, _length) {}
 
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
 private:
@@ -59,6 +63,7 @@ struct ECLRTL_API RtlIntTypeInfo : public RtlTypeInfoBase
 {
     inline RtlIntTypeInfo(unsigned _fieldType, unsigned _length) : RtlTypeInfoBase(_fieldType, _length) {}
 
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
 };
@@ -67,6 +72,7 @@ struct ECLRTL_API RtlSwapIntTypeInfo : public RtlTypeInfoBase
 {
     inline RtlSwapIntTypeInfo(unsigned _fieldType, unsigned _length) : RtlTypeInfoBase(_fieldType, _length) {}
 
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
 };
@@ -76,6 +82,7 @@ struct ECLRTL_API RtlPackedIntTypeInfo : public RtlTypeInfoBase
     inline RtlPackedIntTypeInfo(unsigned _fieldType, unsigned _length) : RtlTypeInfoBase(_fieldType, _length) {}
 
     virtual size32_t size(const byte * self, const byte * selfrow) const;
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
 };
@@ -85,6 +92,7 @@ struct ECLRTL_API RtlStringTypeInfo : public RtlTypeInfoBase
     inline RtlStringTypeInfo(unsigned _fieldType, unsigned _length) : RtlTypeInfoBase(_fieldType, _length) {}
 
     virtual size32_t size(const byte * self, const byte * selfrow) const;
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
 };
@@ -94,6 +102,7 @@ struct ECLRTL_API RtlDataTypeInfo : public RtlTypeInfoBase
     inline RtlDataTypeInfo(unsigned _fieldType, unsigned _length) : RtlTypeInfoBase(_fieldType, _length) {}
 
     virtual size32_t size(const byte * self, const byte * selfrow) const;
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
 };
@@ -103,6 +112,7 @@ struct ECLRTL_API RtlVarStringTypeInfo : public RtlTypeInfoBase
     inline RtlVarStringTypeInfo(unsigned _fieldType, unsigned _length) : RtlTypeInfoBase(_fieldType, _length) {}
 
     virtual size32_t size(const byte * self, const byte * selfrow) const;
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
 };
@@ -112,6 +122,7 @@ struct ECLRTL_API RtlQStringTypeInfo : public RtlTypeInfoBase
     inline RtlQStringTypeInfo(unsigned _fieldType, unsigned _length) : RtlTypeInfoBase(_fieldType, _length) {}
 
     virtual size32_t size(const byte * self, const byte * selfrow) const;
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
 };
@@ -121,6 +132,7 @@ struct ECLRTL_API RtlDecimalTypeInfo : public RtlTypeInfoBase
     inline RtlDecimalTypeInfo(unsigned _fieldType, unsigned _length) : RtlTypeInfoBase(_fieldType, _length) {}
 
     virtual size32_t size(const byte * self, const byte * selfrow) const;
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
 
@@ -131,6 +143,7 @@ struct ECLRTL_API RtlCharTypeInfo : public RtlTypeInfoBase
 {
     inline RtlCharTypeInfo(unsigned _fieldType, unsigned _length) : RtlTypeInfoBase(_fieldType, _length) {}
 
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
 };
@@ -140,6 +153,7 @@ struct ECLRTL_API RtlUnicodeTypeInfo : public RtlTypeInfoBase
 public:
     inline RtlUnicodeTypeInfo(unsigned _fieldType, unsigned _length, const char * _locale) : RtlTypeInfoBase(_fieldType, _length), locale(_locale) {}
 
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
     virtual size32_t size(const byte * self, const byte * selfrow) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
@@ -150,12 +164,12 @@ protected:
     const char * locale;
 };
 
-
 struct ECLRTL_API RtlVarUnicodeTypeInfo : public RtlTypeInfoBase
 {
 public:
     inline RtlVarUnicodeTypeInfo(unsigned _fieldType, unsigned _length, const char * _locale) : RtlTypeInfoBase(_fieldType, _length), locale(_locale) {}
 
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
     virtual size32_t size(const byte * self, const byte * selfrow) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
@@ -166,13 +180,13 @@ protected:
     const char * locale;
 };
 
-
 struct ECLRTL_API RtlUtf8TypeInfo : public RtlTypeInfoBase
 {
 public:
     inline RtlUtf8TypeInfo(unsigned _fieldType, unsigned _length, const char * _locale) : RtlTypeInfoBase(_fieldType, _length), locale(_locale) {}
 
     virtual size32_t size(const byte * self, const byte * selfrow) const;
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
 
@@ -182,7 +196,6 @@ protected:
     const char * locale;
 };
 
-
 struct ECLRTL_API RtlRecordTypeInfo : public RtlTypeInfoBase
 {
     inline RtlRecordTypeInfo(unsigned _fieldType, unsigned _length, const RtlFieldInfo * const * _fields) : RtlTypeInfoBase(_fieldType, _length), fields(_fields) {}
@@ -191,7 +204,7 @@ struct ECLRTL_API RtlRecordTypeInfo : public RtlTypeInfoBase
     virtual size32_t size(const byte * self, const byte * selfrow) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
-
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
     virtual const RtlFieldInfo * const * queryFields() const { return fields; }
 };
 
@@ -209,6 +222,7 @@ struct ECLRTL_API RtlSetTypeInfo : public RtlCompoundTypeInfo
     inline RtlSetTypeInfo(unsigned _fieldType, unsigned _length, const RtlTypeInfo * _child) : RtlCompoundTypeInfo(_fieldType, _length, _child) {}
 
     virtual size32_t size(const byte * self, const byte * selfrow) const;
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
 };
@@ -228,6 +242,7 @@ struct ECLRTL_API RtlDatasetTypeInfo : public RtlCompoundTypeInfo
     inline RtlDatasetTypeInfo(unsigned _fieldType, unsigned _length, const RtlTypeInfo * _child) : RtlCompoundTypeInfo(_fieldType, _length, _child) {}
 
     virtual size32_t size(const byte * self, const byte * selfrow) const;
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
 };
@@ -238,6 +253,7 @@ struct ECLRTL_API RtlDictionaryTypeInfo : public RtlCompoundTypeInfo
     inline RtlDictionaryTypeInfo(unsigned _fieldType, unsigned _length, const RtlTypeInfo * _child) : RtlCompoundTypeInfo(_fieldType, _length, _child) {}
 
     virtual size32_t size(const byte * self, const byte * selfrow) const;
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const;
     virtual size32_t process(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IFieldProcessor & target) const;
     virtual size32_t toXML(const byte * self, const byte * selfrow, const RtlFieldInfo * field, IXmlWriter & target) const;
 };

rtl/eclrtl/rtlint.cpp

@@ -323,11 +323,21 @@ unsigned __int64 rtlReadSwapUInt8(const void * _data)
 }
 
 //no sign extension issues, so same functions can be used for signed and unsigned.
+void rtlWriteSwapInt2(void * _data, unsigned value)
+{
+    _cpyrev2(_data, &value);
+}
+
 void rtlWriteSwapInt3(void * _data, unsigned value)                     
 { 
     _cpyrev3(_data, &value);
 }
 
+void rtlWriteSwapInt4(void * _data, unsigned value)
+{
+    _cpyrev4(_data, &value);
+}
+
 void rtlWriteSwapInt5(void * _data, unsigned __int64 value)     
 { 
     _cpyrev5(_data, &value);
@@ -343,6 +353,12 @@ void rtlWriteSwapInt7(void * _data, unsigned __int64 value)
     _cpyrev7(_data, &value);
 }
 
+void rtlWriteSwapInt8(void * _data, unsigned __int64 value)
+{
+    _cpyrev8(_data, &value);
+}
+
+
 //The following functions are identical to the rtlReadSwapIntX functions for little endian.
 //big endian functions would be different.
 short rtlRevInt2(const void * _data)
@@ -579,4 +595,22 @@ void rtlWriteInt(void * self, __int64 val, unsigned length)
     }
 }
 
+void rtlWriteSwapInt(void * self, __int64 val, unsigned length)
+{
+    switch (length)
+    {
+    case 1: rtlWriteSwapInt1(self, val); break;
+    case 2: rtlWriteSwapInt2(self, val); break;
+    case 3: rtlWriteSwapInt3(self, val); break;
+    case 4: rtlWriteSwapInt4(self, val); break;
+    case 5: rtlWriteSwapInt5(self, val); break;
+    case 6: rtlWriteSwapInt6(self, val); break;
+    case 7: rtlWriteSwapInt7(self, val); break;
+    case 8: rtlWriteSwapInt8(self, val); break;
+    default:
+        rtlFailUnexpected();
+        break;
+    }
+}
+
 #endif

rtl/include/eclhelper.hpp

@@ -62,6 +62,7 @@ interface IException;
 class MemoryBuffer;
 class StringBuffer;
 class rtlRowBuilder;
+class Decimal;
 struct RtlFieldInfo;
 struct RtlTypeInfo;
 
@@ -199,7 +200,33 @@ public:
     virtual void processEndRow(const RtlFieldInfo * field) = 0;
 };
 
-// Functions for processing rows - creating, serializing, destorying etc.
+class RtlDynamicRowBuilder;
+
+interface IFieldSource : public IInterface
+{
+public:
+    virtual bool getBooleanResult(const RtlFieldInfo *field) = 0;
+    virtual void getDataResult(const RtlFieldInfo *field, size32_t &len, void * &result) = 0;
+    virtual double getRealResult(const RtlFieldInfo *field) = 0;
+    virtual __int64 getSignedResult(const RtlFieldInfo *field) = 0;
+    virtual unsigned __int64 getUnsignedResult(const RtlFieldInfo *field) = 0;
+    virtual void getStringResult(const RtlFieldInfo *field, size32_t &len, char * &result) = 0;
+    virtual void getUTF8Result(const RtlFieldInfo *field, size32_t &chars, char * &result) = 0;
+    virtual void getUnicodeResult(const RtlFieldInfo *field, size32_t &chars, UChar * &result) = 0;
+    virtual void getDecimalResult(const RtlFieldInfo *field, Decimal &value) = 0;
+
+    //The following are used process the structured fields
+    virtual void processBeginSet(const RtlFieldInfo * field, bool &isAll) = 0;
+    virtual void processBeginDataset(const RtlFieldInfo * field) = 0;
+    virtual void processBeginRow(const RtlFieldInfo * field) = 0;
+    virtual bool processNextSet(const RtlFieldInfo * field) = 0;
+    virtual bool processNextRow(const RtlFieldInfo * field) = 0;
+    virtual void processEndSet(const RtlFieldInfo * field) = 0;
+    virtual void processEndDataset(const RtlFieldInfo * field) = 0;
+    virtual void processEndRow(const RtlFieldInfo * field) = 0;
+};
+
+// Functions for processing rows - creating, serializing, destroying etc.
 interface IOutputRowSerializer;
 interface IOutputRowDeserializer;
 
@@ -313,6 +340,8 @@ interface RtlITypeInfo
     virtual const char * queryLocale() const = 0;
     virtual const RtlFieldInfo * const * queryFields() const = 0;               // null terminated list
     virtual const RtlTypeInfo * queryChildType() const = 0;
+
+    virtual size32_t build(ARowBuilder &builder, size32_t offset, const RtlFieldInfo *field, IFieldSource &source) const = 0;
 };
 
 
@@ -340,6 +369,7 @@ public:
                                     // if RFTMunknownsize then maxlength (records) [maxcount(datasets)]
 };
 
+
 //Core struct used for representing meta for a field.
 struct RtlFieldInfo
 {
@@ -357,6 +387,10 @@ struct RtlFieldInfo
     { 
         return type->size(self, selfrow); 
     }
+    inline size32_t build(ARowBuilder &builder, size32_t offset, IFieldSource & source) const
+    {
+        return type->build(builder, offset, this, source);
+    }
     inline size32_t process(const byte * self, const byte * selfrow, IFieldProcessor & target) const 
     {
         return type->process(self, selfrow, this, target);

testing/ecl/streame.ecl

@@ -17,8 +17,23 @@
 
 IMPORT Python;
 
+childrec := RECORD
+   string name => unsigned value;
+END;
 namesRecord := RECORD
-    STRING name;
+    STRING name1;
+    STRING10 name2;
+    DATASET(childrec) childnames;
+    DICTIONARY(childrec) childdict;
+    unsigned1 val1;
+    integer1   val2;
+    UTF8 u1;
+    UNICODE u2;
+    UNICODE8 u3;
+    BIG_ENDIAN unsigned6 val3;
+    DATA d;
+    BOOLEAN b;
+    SET OF STRING ss1;
 END;
 
 dataset(namesRecord) blockedNames(string prefix) := EMBED(Python)
@@ -29,13 +44,10 @@ _linkcounted_ dataset(namesRecord) linkedNames(string prefix) := EMBED(Python)
   return ["Gavin","John","Bart"]
 ENDEMBED;
 
-streamed dataset(namesRecord) streamedNames(string prefix) := EMBED(Python)
-  return ["Gavin","John","Bart"]
+STREAMED dataset(namesRecord) streamedNames(data d, utf8 u) := EMBED(Python)
+  return [  \
+     ("Gavin", "Halliday", [("a", 1)], 250, -1,  U'là',  U'là',  U'là', 1234566, d, False, {"1","2"}), \
+     ("John", "Smith", [], 250, -1,  U'là',  U'là',  u, 1234566, d, True, [])]
 ENDEMBED;
 
-titles := dataset(['', 'Mr. ', 'Rev. '], { string title });
-
-//output(normalize(titles, blockedNames(left.title), transform(right)));
-//output(normalize(titles, linkedNames(left.title), transform(right)));
-//output(normalize(titles, streamedNames(left.title), transform(right)));
-output(streamedNames('mr'));
+output(streamedNames(d'AA', u'là'));