HPCC-22129 Can't pass datasets via iterator to embedded Java

Various other iterator-related issues fixed besides the one mentioned in the
title.

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

plugins/javaembed/HpccClassLoader.java

@@ -71,16 +71,17 @@ public class HpccClassLoader extends java.lang.ClassLoader
     }
     public synchronized Class<?> findClass(String className) throws ClassNotFoundException
     {
-        Class<?> result = classes.get(className);
+        String luName = className.replace(".","/");
+        Class<?> result = classes.get(luName);
         if (result == null)
         {
             if (bytecodeLen != 0)
-                result = defineClassForEmbed(bytecodeLen, bytecode, className.replace(".","/"));
+                result = defineClassForEmbed(bytecodeLen, bytecode, luName);
             if ( result == null && pathLoader != null)
                 result = pathLoader.loadClass(className);
             if (result == null)
                 return super.findClass(className);
-            classes.put(className, result);
+            classes.put(luName, result);
         }
         return result; 
     }

plugins/javaembed/javaembed.cpp

@@ -363,7 +363,9 @@ public:
     jobject ToReflectedField(jclass cls, jfieldID fieldID, jboolean isStatic) {
         return checkException(functions->ToReflectedField(this,cls,fieldID,isStatic));
     }
-
+    jboolean IsAssignableFrom(jclass clazz1, jclass clazz2) {
+        return checkException(functions->IsAssignableFrom(this,clazz1,clazz2));
+    }
     jclass GetObjectClass(jobject obj)
     {
         return checkException(JNIEnv::GetObjectClass(obj));
@@ -650,6 +652,7 @@ static jclass customLoaderClass;
 static jmethodID clc_newInstance;
 static jmethodID clc_getSignature;
 static jclass hpccIteratorClass;
+static jclass utilIteratorClass;
 static jmethodID hi_constructor;
 
 static jclass systemClass;
@@ -706,6 +709,7 @@ static void setupGlobals(CheckedJNIEnv *J)
         arrayList_toArray = J->GetMethodID(arrayListClass, "toArray", "()[Ljava/lang/Object;" );
 
         langStringClass = J->FindGlobalClass("java/lang/String");
+        utilIteratorClass = J->FindGlobalClass("java/util/Iterator");
 
         langIllegalArgumentExceptionClass = J->FindGlobalClass("java/lang/IllegalArgumentException");
     }
@@ -1809,14 +1813,13 @@ public:
       javaBuilder(JNIenv, &dummyField, className)
     {
         nextRead = false;
-        nextPending = NULL;
     }
 
     bool hasNext()
     {
         if (!nextRead)
         {
-            nextPending = (const byte *) val->ungroupedNextRow();
+            nextPending.setown(val->ungroupedNextRow());
             nextRead = true;
             if (!nextPending)
                 val->stop();
@@ -1828,7 +1831,7 @@ public:
     {
         if (!hasNext())
             return NULL;
-        typeInfo->process(nextPending, nextPending, &dummyField, javaBuilder); // Creates a java object from the incoming ECL row
+        typeInfo->process((const byte *) nextPending.get(), (const byte *) nextPending.get(), &dummyField, javaBuilder); // Creates a java object from the incoming ECL row
         nextRead = false;
         return javaBuilder.getObject();
     }
@@ -1837,7 +1840,7 @@ protected:
     Linked<IRowStream> val;
     RtlFieldStrInfo dummyField;
     JavaObjectBuilder javaBuilder;
-    const byte *nextPending;
+    roxiemem::OwnedConstRoxieRow nextPending;
     bool nextRead;
 };
 
@@ -1867,8 +1870,12 @@ public:
     JavaRowStream(jobject _iterator, IEngineRowAllocator *_resultAllocator)
     : resultAllocator(_resultAllocator)
     {
-        iterator = queryJNIEnv()->NewGlobalRef(_iterator, "iterator");
-        // Note that we can't cache the JNIEnv, iterClass, or methodIds here - calls may be made on different threads (though not at the same time).
+        CheckedJNIEnv *JNIenv = queryJNIEnv();
+        iterator = JNIenv->NewGlobalRef(_iterator, "iterator");
+        iterClass = (jclass) JNIenv->NewGlobalRef(JNIenv->GetObjectClass(iterator), "iterClass");
+        hasNextMethod = JNIenv->GetMethodID(iterClass, "hasNext", "()Z" );
+        nextMethod = JNIenv->GetMethodID(iterClass, "next", "()Ljava/lang/Object;" );
+        // Note that we can't save the JNIEnv value here - calls may be made on different threads (though not at the same time).
     }
     ~JavaRowStream()
     {
@@ -1887,15 +1894,12 @@ public:
         //    return NULL;
         // }
         // result = iterator.next();
-        jclass iterClass =JNIenv->GetObjectClass(iterator);
-        jmethodID hasNextMethod = JNIenv->GetMethodID(iterClass, "hasNext", "()Z" );
         jboolean hasNext = JNIenv->CallBooleanMethod(iterator, hasNextMethod);
         if (!hasNext)
         {
             stop();
             return NULL;
         }
-        jmethodID nextMethod = JNIenv->GetMethodID(iterClass, "next", "()Ljava/lang/Object;" );
         jobject result = JNIenv->CallObjectMethod(iterator, nextMethod);
         RtlDynamicRowBuilder rowBuilder(resultAllocator);
         const RtlTypeInfo *typeInfo = resultAllocator->queryOutputMeta()->queryTypeInfo();
@@ -1916,12 +1920,22 @@ public:
                 JNIenv->DeleteGlobalRef(iterator);
                 iterator = NULL;
             }
+            if (iterClass)
+            {
+                JNIenv->DeleteGlobalRef(iterClass);
+                iterClass = NULL;
+            }
+            hasNextMethod = nullptr;
+            nextMethod = nullptr;
         }
     }
 
 protected:
     Linked<IEngineRowAllocator> resultAllocator;
-    jobject iterator;
+    jobject iterator = nullptr;
+    jclass iterClass = nullptr;
+    jmethodID hasNextMethod = nullptr;
+    jmethodID nextMethod = nullptr;
 };
 
 const char *esdl2JavaSig(IEsdlDefinition &esdl, const char *esdlType)
@@ -2874,10 +2888,27 @@ public:
             b.read(attr_count);
             for (unsigned j = 0; j < attr_count; j++)
             {
-                b.skip(2);
+                uint16_t attr_name_idx; b.read(attr_name_idx);
+                StringAttr attrName;
+                readUtf(attrName, attr_name_idx);
                 uint32_t attr_length;
                 b.read(attr_length);
-                b.skip(attr_length);
+                if (streq(attrName, "Signature") && attr_length==2)
+                {
+                    uint16_t ext_sig_idx; b.read(ext_sig_idx);
+                    StringAttr extSig;
+                    readUtf(extSig, ext_sig_idx);
+#ifdef TRACE_CLASSFILE
+                    DBGLOG("Seen extended signature %s", extSig.str());
+#endif
+                    if (flags & ACC_PUBLIC)
+                    {
+                        methodSigs.pop();
+                        methodSigs.append(extSig);
+                    }
+                }
+                else
+                    b.skip(attr_length);
             }
         }
         /* Don't bother reading attributes as they are not really interesting to us
@@ -3050,7 +3081,13 @@ public:
         if (codeCtx)
         {
             engine = codeCtx->queryEngineContext();
-            nodeNum = codeCtx->getNodeNum();
+            try {
+                nodeNum = codeCtx->getNodeNum();
+            }
+            catch (IException *E)
+            {
+                E->Release();  // We may get an error if calling on the master - we just want to ignore it
+            }
         }
         StringBuffer lclassPath;
         if (engine)
@@ -3423,6 +3460,13 @@ public:
     }
     virtual IRowStream *getDatasetResult(IEngineRowAllocator * _resultAllocator)
     {
+        jclass iterClass =JNIenv->GetObjectClass(result.l);
+        if (!JNIenv->IsAssignableFrom(iterClass, utilIteratorClass))
+        {
+            StringBuffer s;
+            throw MakeStringException(0, "javaembed: In method %s: Java code should return an iterator", getReportName(s).str());
+            // MORE - perhaps we should also support Iterable?
+        }
         return new JavaRowStream(result.l, _resultAllocator);
     }
     virtual byte * getRowResult(IEngineRowAllocator * _resultAllocator)
@@ -4075,6 +4119,7 @@ public:
         if (persistMode==persistNone)
             instance = 0;  // otherwise we leave it for next call as it saves a lot of time looking it up
         JNIenv->PopLocalFrame(nullptr);
+        iterators.kill();
 #ifdef FORCE_GC
         forceGC(JNIenv);
 #endif

testing/regress/ecl/java-stream.ecl

@@ -0,0 +1,78 @@
+import java;
+
+/*
+ This example illustrates various calls to Java functions defined in the Java module JavaCat.
+ The source of JavaCat can be found in the examples directory - it can be compiled to JavaCat.class
+ using
+
+   javac JavaCat
+
+ and the resulting file JavaCat.class should be placed in /opt/HPCCSystems/classes (or somewhere else
+ where it can be located via the standard Java CLASSPATH environment variable.
+ */
+
+// Passing and returning records and datasets
+// When passing/returning a record, the corresponding Java function should take/return an object as a parameter whose fields
+// can be mapped by name to the ECL record fields
+
+nrec := record
+  utf8 ufield;
+end;
+
+jret := RECORD
+  boolean bfield;
+  integer4 ifield;
+  integer8 lfield;
+  real8 dfield;
+  real4 ffield;
+  string1 cfield1;
+  string1 cfield2;
+  string sfield;
+  nrec n;
+  set of boolean bset;
+  set of data dset;
+  set of string sset;
+  LINKCOUNTED DATASET(nrec) sub;
+end;
+
+jret jreturnrec(boolean b, integer i, real8 d) := IMPORT(java, 'JavaCat.returnrec:(ZID)LJavaCat;');
+STRING jpassrec(jret r) := IMPORT(java, 'JavaCat.passrec:(LJavaCat;)Ljava/lang/String;');
+
+ret := jreturnrec(false, 10, 2.345);
+
+OUTPUT(ret);            // Calls a Java function that returns an ECL record
+OUTPUT(jpassrec(ret));  // Passes an ECL record to a Java function
+
+// When passing a dataset to a Java function, the Java function should take either an array or an iterator of objects,
+// where the fields of the object in question are mapped by name to the fields in the ECL record.
+// When passing an iterator, we use a modified form of the function signature in the IMPORT statement, using a < to indicate "Iterator of"
+// followed by the name of the class of the objects that the iterator is to return. This is the one case where the output of javap -s cannot be used
+// directly to provide the signature of the java function being called. We can also use the "extended" signature of the method that is output by
+// javap -s -v, of the form 'JavaCat.passDataset:(Ljava/util/Iterator<LJavaCat;>;)I'
+//
+// To return a dataset, an iterator must be returned.
+
+INTEGER passDataset(LINKCOUNTED DATASET(jret) d) :=
+  IMPORT(java, 'JavaCat.passDataset:(<LJavaCat;)I'); // Calls int passDataset(Iterator<JavaCat> d)
+// Note we could also use 'Ljava/util/Iterator<LJavaCat;>;)I' as the signature, but not 'Ljava/util/Iterator;)I' which is the signature output by javap -s
+
+DATASET(jret) passDataset2(LINKCOUNTED DATASET(jret) d) :=
+  IMPORT(java, 'JavaCat.passDataset2:([LJavaCat;)Ljava/util/Iterator;'); // Calls Iterator<JavaCat> passDataset2(JavaCat d[])
+
+ds := DATASET(
+  [
+     {true, 1,2,3,4,'a', 'b', 'cd', u'ef', [true,false], [], ['Hello from ECL'], [{'1'},{'2'},{'3'},{'4'},{'5'}]}
+    ,{true, 2,4,3,4,'a', 'b', 'cd', u'ef', [true,false], [], [], []}
+    ,{true, 3,6,3,4,'a', 'b', 'cd', u'ef', [true,false], [], [], []}
+    ,{true, 8,8,3,4,'a', 'b', 'cd', u'ef', [true,false], [d'AA55'], [], []}
+  ], jret);
+
+output(passDataset(ds));  // Using an iterator
+output(passDataset2(ds)); // using an array, and illustrating the return of a dataset
+
+// It is also possible to code a traonsform function in Java - both the parameter and the return type should be a
+// Java object type that maps the fields of the ECL record by name.
+
+transform(jret) testTransform(jret in, integer lim) := IMPORT(java, 'JavaCat.transform:(LJavaCat;I)LJavaCat;');
+
+output(project(ds, testTransform(LEFT, COUNTER)));

testing/regress/ecl/key/java-stream.xml

@@ -0,0 +1,21 @@
+<Dataset name='Result 1'>
+ <Row><bfield>false</bfield><ifield>10</ifield><lfield>1000000000</lfield><dfield>2.345</dfield><ffield>2.345000028610229</ffield><cfield1>X</cfield1><cfield2>Z</cfield2><sfield>Yoohoo</sfield><n><ufield>nest</ufield></n><bset><Item>false</Item><Item>false</Item><Item>false</Item><Item>false</Item><Item>false</Item></bset><dset><Item>0E</Item></dset><sset><Item>Hello</Item></sset><sub><Row><ufield>subnest</ufield></Row></sub></Row>
+</Dataset>
+<Dataset name='Result 2'>
+ <Row><Result_2>nest</Result_2></Row>
+</Dataset>
+<Dataset name='Result 3'>
+ <Row><Result_3>20</Result_3></Row>
+</Dataset>
+<Dataset name='Result 4'>
+ <Row><bfield>true</bfield><ifield>1</ifield><lfield>2</lfield><dfield>3.0</dfield><ffield>4.0</ffield><cfield1>a</cfield1><cfield2>b</cfield2><sfield>cd</sfield><n><ufield>ef</ufield></n><bset><Item>true</Item><Item>false</Item></bset><dset></dset><sset><Item>Hello from ECL</Item></sset><sub><Row><ufield>1</ufield></Row><Row><ufield>2</ufield></Row><Row><ufield>3</ufield></Row><Row><ufield>4</ufield></Row><Row><ufield>5</ufield></Row></sub></Row>
+ <Row><bfield>true</bfield><ifield>2</ifield><lfield>4</lfield><dfield>3.0</dfield><ffield>4.0</ffield><cfield1>a</cfield1><cfield2>b</cfield2><sfield>cd</sfield><n><ufield>ef</ufield></n><bset><Item>true</Item><Item>false</Item></bset><dset></dset><sset></sset><sub></sub></Row>
+ <Row><bfield>true</bfield><ifield>3</ifield><lfield>6</lfield><dfield>3.0</dfield><ffield>4.0</ffield><cfield1>a</cfield1><cfield2>b</cfield2><sfield>cd</sfield><n><ufield>ef</ufield></n><bset><Item>true</Item><Item>false</Item></bset><dset></dset><sset></sset><sub></sub></Row>
+ <Row><bfield>true</bfield><ifield>8</ifield><lfield>8</lfield><dfield>3.0</dfield><ffield>4.0</ffield><cfield1>a</cfield1><cfield2>b</cfield2><sfield>cd</sfield><n><ufield>ef</ufield></n><bset><Item>true</Item><Item>false</Item></bset><dset><Item>41413535</Item></dset><sset></sset><sub></sub></Row>
+</Dataset>
+<Dataset name='Result 5'>
+ <Row><bfield>true</bfield><ifield>1</ifield><lfield>100000000</lfield><dfield>3.0</dfield><ffield>3.0</ffield><cfield1>X</cfield1><cfield2>Z</cfield2><sfield>Yoohoo</sfield><n><ufield>nest</ufield></n><bset><Item>false</Item><Item>false</Item><Item>false</Item><Item>true</Item><Item>false</Item></bset><dset><Item>0E</Item></dset><sset><Item>Hello</Item></sset><sub><Row><ufield>subnest</ufield></Row></sub></Row>
+ <Row><bfield>true</bfield><ifield>2</ifield><lfield>200000000</lfield><dfield>3.0</dfield><ffield>3.0</ffield><cfield1>X</cfield1><cfield2>Z</cfield2><sfield>Yoohoo</sfield><n><ufield>nest</ufield></n><bset><Item>false</Item><Item>false</Item><Item>false</Item><Item>true</Item><Item>false</Item></bset><dset><Item>0E</Item></dset><sset><Item>Hello</Item></sset><sub><Row><ufield>subnest</ufield></Row></sub></Row>
+ <Row><bfield>true</bfield><ifield>3</ifield><lfield>300000000</lfield><dfield>3.0</dfield><ffield>3.0</ffield><cfield1>X</cfield1><cfield2>Z</cfield2><sfield>Yoohoo</sfield><n><ufield>nest</ufield></n><bset><Item>false</Item><Item>false</Item><Item>false</Item><Item>true</Item><Item>false</Item></bset><dset><Item>0E</Item></dset><sset><Item>Hello</Item></sset><sub><Row><ufield>subnest</ufield></Row></sub></Row>
+ <Row><bfield>true</bfield><ifield>4</ifield><lfield>400000000</lfield><dfield>3.0</dfield><ffield>3.0</ffield><cfield1>X</cfield1><cfield2>Z</cfield2><sfield>Yoohoo</sfield><n><ufield>nest</ufield></n><bset><Item>false</Item><Item>false</Item><Item>false</Item><Item>true</Item><Item>false</Item></bset><dset><Item>0E</Item></dset><sset><Item>Hello</Item></sset><sub><Row><ufield>subnest</ufield></Row></sub></Row>
+</Dataset>