class VespaDocumentOperation extends EvalFunc<String> {

    @Override
    public String exec(Tuple tuple) throws IOException {
        if (tuple == null || tuple.size() == 0) {
            if (statusReporter != null) {
                statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation failed", 1);
            }
            return null;
        }
        if (template == null || template.length() == 0) {
            if (statusReporter != null) {
                statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation failed", 1);
            }
            warnLog("No valid document id template found. Skipping.", PigWarning.UDF_WARNING_1);
            return null;
        }
        if (operation == null) {
            warnLog("No valid operation found. Skipping.", PigWarning.UDF_WARNING_2);
            return null;
        }

        String json = null;

        try {
            if (reporter != null) {
                reporter.progress();
            }

            Schema inputSchema = getInputSchema();
            Map<String, Object> fields = TupleTools.tupleMap(inputSchema, tuple);
            String docId = TupleTools.toString(fields, template);
            if (verbose) {
                System.out.println("Processing docId: "+ docId);
            }
            // create json
            json = create(operation, docId, fields, properties, inputSchema);
            if (json == null || json.length() == 0) {
                warnLog("No valid document operation could be created.", PigWarning.UDF_WARNING_3);
                return null;
            }


        } catch (Exception e) {
            if (statusReporter != null) {
                statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation failed", 1);
            }
            StringBuilder sb = new StringBuilder();
            sb.append("Caught exception processing input row: \n");
            sb.append(tuple.toString());
            sb.append("\nException: ");
            sb.append(getStackTraceAsString(e));
            warnLog(sb.toString(), PigWarning.UDF_WARNING_4);
            return null;
        }
        if (statusReporter != null) {
            statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation ok", 1);
        }
        return json;
    }

    public  VespaDocumentOperation(String... params);

}