AnalyticDB:UDFs

public class App
        implements StreamRequestHandler, FunctionInitializer
{
    public static final Logger log = Logger.getLogger(App.class.getName());

    public void initialize(Context context) throws IOException {
        // TODO
    }

    @Override
    public void handleRequest(
            InputStream inputStream, OutputStream outputStream, Context context)
            throws IOException
    {
        // Step 1: Decompress the input.
        // REMOTE_CALL() compresses data with GZIP by default (compressed=true).
        // If you set compressed=false in external_config, skip this step.
        InputStream decompressedInput = tryUnCompress(inputStream);

        JSONObject response = new JSONObject();
        try {
            // Step 2: Parse the JSON body sent by AnalyticDB for MySQL.
            // The body has the structure: {"rowCount": N, "compressed": true/false, "data": [[...], ...]}
            JSONObject requestJson = JSONObject.parseObject(IOUtils.toString(decompressedInput));
            if (requestJson.containsKey("data")) {
                JSONArray result = new JSONArray();
                // Step 3: Iterate over the rows in the "data" array.
                // Each row is a JSON array whose columns match the X1...Xn parameter order in REMOTE_CALL().
                // This function expects two integers: X1 (index 0) and X2 (index 1).
                JSONArray data = requestJson.getJSONArray("data");
                for (int i = 0; i < data.size(); i++) {
                    JSONArray row = data.getJSONArray(i);
                    if (row.size() == 2) {
                        result.add(testFunc(row.getInteger(0), row.getInteger(1)));
                    } else {
                        throw new RuntimeException("row size is not 2");
                    }
                }
                // Step 4: Build the response.
                response.put("result", result);
                response.put("success", true);
                response.put("message", "");
            } else {
                response.put("success", false);
                response.put("message", "no data inside");
            }
        } catch (Exception e) {
            log.info("error happened" + e.getMessage());
            response.put("success", false);
            response.put("message", e.getMessage());
        }

        // Step 5: Compress the response.
        // REMOTE_CALL() expects a GZIP-compressed response when compressed=true (the default).
        // If you set compressed=false in external_config, remove this compression step.
        outputStream.write(tryCompress(response.toJSONString().getBytes()));
    }

    private String testFunc(int a, int b) {
        // Concatenate a, b, and 1 with ampersands.
        return String.valueOf(a) + '&' + b + '&' + 1;
    }

    public static byte[] tryCompress(byte[] bytes) {
        ByteArrayOutputStream byteOutputStream = new ByteArrayOutputStream();
        try {
            GZIPOutputStream gzipOutputStream = new GZIPOutputStream(byteOutputStream);
            gzipOutputStream.write(bytes);
            gzipOutputStream.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return byteOutputStream.toByteArray();
    }

    public static InputStream tryUnCompress(InputStream inputStream) throws IOException {
        return new GZIPInputStream(inputStream);
    }
}

-- Concatenate 1 and 2 with an ampersand
SELECT remote_call(
  'varchar',
  'ConcactNumberWithCompress',
  '{endpoint:"1234567890000****.cn-zhangjiakou-internal.fc.aliyuncs.com"}',
  1, 2
);
-- Result: 1&2&1

SELECT remote_call(
  'varchar',
  'ConcactNumberWithCompress',
  '{endpoint:"1234567890000****.cn-zhangjiakou-internal.fc.aliyuncs.com",compressed:false,max_batch_size:5000000}',
  3, 4
);
-- Result: 3&4&1

SELECT remote_call('varchar', 'ConcactNumberWithCompress', null, 5, 6);
-- Result: 5&6&1