HPCC-10457 Streamed dataset support for Python

Initial row and transform support

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

plugins/pyembed/CMakeLists.txt

@@ -24,6 +24,7 @@
 #####################################################
 
 set ( debug_python Off )   # A lot slower but can assist in debugging...
+SET ( DEBUG_PYTHON_LIBRARY "/usr/lib/libpython2.7_d.so" )
 
 project( pyembed )
 
@@ -58,7 +59,7 @@ if (USE_PYTHON)
 
     install ( TARGETS pyembed DESTINATION plugins )
     if (debug_python)
-      target_link_libraries ( pyembed ${PYTHON_DEBUG_LIBRARY} )
+      target_link_libraries ( pyembed ${DEBUG_PYTHON_LIBRARY} )
     else()
       target_link_libraries ( pyembed ${PYTHON_LIBRARY} )
     endif()

plugins/pyembed/pyembed.cpp

@@ -581,8 +581,9 @@ class PythonRowBuilder : public CInterfaceOf<IFieldSource>
 {
 public:
     PythonRowBuilder(PyObject *_row)
-    : iter(NULL), elem(NULL), pushback(_row), named(false)
+    : iter(NULL), elem(NULL), named(false)
     {
+        pushback.set(_row);
     }
     virtual bool getBooleanResult(const RtlFieldInfo *field)
     {
@@ -705,6 +706,7 @@ protected:
         parent.set(elem);
         iter.setown(PyObject_GetIter(elem));
         named = isNamedTuple(elem);
+        elem.clear();
     }
     bool isNamedTuple(PyObject *obj)
     {
@@ -734,6 +736,15 @@ protected:
     BoolArray namedStack;
 };
 
+static size32_t getRowResult(PyObject *result, ARowBuilder &builder)
+{
+    PythonRowBuilder pyRowBuilder(result);
+    const RtlTypeInfo *typeInfo = builder.queryAllocator()->queryOutputMeta()->queryTypeInfo();
+    assertex(typeInfo);
+    RtlFieldStrInfo dummyField("<row>", NULL, typeInfo);
+    return typeInfo->build(builder, 0, &dummyField, pyRowBuilder);
+}
+
 // GILBlock ensures the we hold the Python "Global interpreter lock" for the appropriate duration
 
 class GILBlock
@@ -778,11 +789,7 @@ public:
         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);
+        size32_t len = pyembed::getRowResult(row, rowBuilder);
         return rowBuilder.finalizeRowClear(len);
     }
     virtual void stop()
@@ -864,11 +871,13 @@ public:
     }
     virtual byte * getRowResult(IEngineRowAllocator * _resultAllocator)
     {
-        UNIMPLEMENTED;
+        RtlDynamicRowBuilder rowBuilder(_resultAllocator);
+        size32_t len = pyembed::getRowResult(result, rowBuilder);
+        return (byte *) rowBuilder.finalizeRowClear(len);
     }
     virtual size32_t getTransformResult(ARowBuilder & builder)
     {
-        UNIMPLEMENTED;
+        return pyembed::getRowResult(result, builder);
     }
     virtual void bindBooleanParam(const char *name, bool val)
     {

testing/ecl/key/streame2.xml

@@ -0,0 +1,32 @@
+<Dataset name='Result 1'>
+ <Row><name>Generate:</name><value>0</value></Row>
+ <Row><name>Generate:</name><value>1</value></Row>
+ <Row><name>Generate:</name><value>2</value></Row>
+ <Row><name>Generate:</name><value>3</value></Row>
+ <Row><name>Generate:</name><value>4</value></Row>
+ <Row><name>Generate:</name><value>5</value></Row>
+ <Row><name>Generate:</name><value>6</value></Row>
+ <Row><name>Generate:</name><value>7</value></Row>
+ <Row><name>Generate:</name><value>8</value></Row>
+ <Row><name>Generate:</name><value>9</value></Row>
+</Dataset>
+<Dataset name='Result 2'>
+ <Row><name>name1</name><value>1</value></Row>
+ <Row><name>name2</name><value>2</value></Row>
+</Dataset>
+<Dataset name='Result 3'>
+ <Row><name>1</name></Row>
+ <Row><name>2</name></Row>
+ <Row><name>3</name></Row>
+</Dataset>
+<Dataset name='Result 4'>
+ <Row><name>1</name></Row>
+ <Row><name>2</name></Row>
+ <Row><name>3</name></Row>
+</Dataset>
+<Dataset name='Result 5'>
+ <Row><name>Hello</name><value>10</value></Row>
+</Dataset>
+<Dataset name='Result 6'>
+ <Row><name>Hello</name><value>10</value></Row>
+</Dataset>

testing/ecl/streame.ecl

@@ -63,4 +63,4 @@ dataset(childrec) testGenerator(unsigned lim) := EMBED(Python)
 ENDEMBED;
 
 output(streamedNames(d'AA', u'là'));
-output (testGenerator(10));
+output (testGenerator(10));

testing/ecl/streame2.ecl

@@ -65,10 +65,10 @@ transform(childrec) testTransform(unsigned lim) := EMBED(Python)
 ENDEMBED;
 
 
-//output (testGenerator(10));
-//output (testNamedTuple(10));
-//output (testMissingTuple1(10));
-//output (testMissingTuple2(10));
+output (testGenerator(10));
+output (testNamedTuple(10));
+output (testMissingTuple1(10));
+output (testMissingTuple2(10));
 
 output(testRowReturn(10));
-//output(row(testTransform(10)));
+output(row(testTransform(10)));