Java language scalar function

Code

The code in this example shows that, with minor changes, this program works in remote mode. The change is in the main where a loop is added around getAPI so that it gets called two times. If the connection is remote, it is handed off to be run by a new thread and to wait for a new connection. Finally, if the connection is not remote, break out of the loop after the first iteration. This code works like a local AE and handles two iterations as a remote AE.

import org.netezza.ae.*;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class TestJavaInterface {

    private static final Executor exec =
        Executors.newCachedThreadPool();

    public static final void main(String [] args) {
        try {
            mainImpl(args);
        } catch (Throwable t) {
            System.err.println(t.toString());
            NzaeUtil.logException(t, "main");
        }
    }

    public static final void mainImpl(String [] args) {
        NzaeApiGenerator helper = new NzaeApiGenerator();
 
       while (true) {
           final NzaeApi api = helper.getApi(NzaeApi.FUNCTION);
           if (api.apiType == NzaeApi.FUNCTION) {
               if (!helper.isRemote()) {
                   run(api.aeFunction);
                   break;
               } else {
                   Runnable task = new Runnable() {
                           public void run() {
                               try {
                                   TestJavaInterface.run(api.aeFunction);
                               } finally {
                                   api.aeFunction.close();
                               }
                           }
                       };
                   exec.execute(task);
               }
           }
        }
        helper.close();
    }

    public static class MyHandler implements NzaeMessageHandler
    {
        public void evaluate(Nzae ae, NzaeRecord input, NzaeRecord output) { 
            final NzaeMetadata meta = ae.getMetadata();
  
            int op = 0;
            double result = 0;

            if (meta.getOutputColumnCount() != 1 ||
                meta.getOutputNzType(0) != NzaeDataTypes.NZUDSUDX_DOUBLE) {
                throw new NzaeException("expecting one output column of type
double");

            }
            if (meta.getInputColumnCount() < 1) {
                throw new NzaeException("expecting at least one input column");
            }
            if (meta.getInputNzType(0) != NzaeDataTypes.NZUDSUDX_FIXED &&
                meta.getInputNzType(0) != NzaeDataTypes.NZUDSUDX_VARIABLE) {
                throw new NzaeException("first input column expected to be a
string type");
            }

            for (int i = 0; i < input.size(); i++) {
                if (input.getField(i) == null) {
                continue;
            }
            int dataType = meta.getInputNzType(i);
            if (i == 0) {
                if (!(dataType == NzaeDataTypes.NZUDSUDX_FIXED
                        || dataType == NzaeDataTypes.NZUDSUDX_VARIABLE)) {
 
                    ae.userError("first column must be a string");
                }
                String opStr = input.getFieldAsString(0);
                if (opStr.equals("*")) {
                    result = 1;
                    op = OP_MULT;
                }
                else if (opStr.equals("+")) {
                    result = 0;
                    op = OP_ADD;
                }
                else {
                    ae.userError("invalid operator = " + opStr);
                }
                continue;
            }
            switch (dataType) {
                case NzaeDataTypes.NZUDSUDX_INT8:
                case NzaeDataTypes.NZUDSUDX_INT16:
                case NzaeDataTypes.NZUDSUDX_INT32:
                case NzaeDataTypes.NZUDSUDX_INT64:
                case NzaeDataTypes.NZUDSUDX_FLOAT:
                case NzaeDataTypes.NZUDSUDX_DOUBLE:
                case NzaeDataTypes.NZUDSUDX_NUMERIC32:
                case NzaeDataTypes.NZUDSUDX_NUMERIC64:
                case NzaeDataTypes.NZUDSUDX_NUMERIC128:
                    switch (op) {
                        case OP_ADD:
                            result +=
input.getFieldAsNumber(i).doubleValue();
                            break;
                        case OP_MULT:
                            result *=
input.getFieldAsNumber(i).doubleValue();
                            break;
                    default:
                        break;
                    }
                    break;
                default:
                    break;
            }
        }   // end of column for loop

        output.setField(0, result);
        }

     }
     private static final int OP_ADD = 1;
     private static final int OP_MULT = 2;

     public static int run(Nzae ae)
     {
         ae.run(new MyHandler());
         return 0;
     }
}

Compilation

Use the standard compile as in local mode:

$NZ_EXPORT_DIR/ae/utilities/bin/compile_ae --language java --template \
     compile TestJavaInterface.java --version 3

Registration

Registration is slightly different. First run this to register:

$NZ_EXPORT_DIR/ae/utilities/bin/register_ae --sig "rem_applyop_java(varargs)" \
     --return "double" --class AeUdf --language java --template udf \
     --define "java_class=TestJavaInterface" --version 3 --remote \
     --rname testjavapi

This specifies the code be registered as a remote ae with a remote name of testjavapi, which matches the code. In addition, a launcher is registered:

$NZ_EXPORT_DIR/ae/utilities/bin/register_ae --sig "rem_applyop_launch(int8)" \
     --return "TABLE(aeresult varchar(255))" --class AeUdtf --language java \
     --template udtf --define "java_class=TestJavaInterface" --version 3 \
     --remote --rname testjavapi --launch

With the exception of the --rname, --exe and the name portion of --sig, all launchers look like the above. In this case it is a UDTF, since that is the interface for all launchers. The value of --rname must match the name in the code.

Running

To run the AE in remote mode, the executable is run as a “server.” In this instance it handles only queries run on the host. Usually, AEs are started on the SPUs as well. Start an instance on the host:

SELECT * FROM TABLE WITH FINAL(rem_applyop_launch(0));
                                AERESULT
-------------------------------------------------------------------------
 tran: 14896 session: 18286 DATA slc: 0 hardware: 0 machine: bdrosendev process:
15937 thread: 15938
(1 row)

Notice that a different syntax is used to invoke the table function style launcher. This is the syntax used to call any UDTF-based AE. Now run the AE:

SELECT rem_applyop_java('+', 4,5,1.1);
 REM_APPLYOP_JAVA
------------------
 10.1
(1 row)

Finally, cause an error:

SELECT rem_applyop_java(1);
ERROR: first input column expected to be a string type