@Override
    public Object parse(SourceData sourceData, int rowIndex, Context context) {
        Object stringObject = stringParser.parse(sourceData, rowIndex, context);
        if (stringObject == null) {
            return null;
        }
        String str = OperatorTools.parseString(stringObject);
        if (str.isEmpty()) {
            return "";
        }

        // parse compress type
        String compressType = DEFAULT_COMPRESS_TYPE;
        if (compressTypeParser != null) {
            Object compressTypeObj = compressTypeParser.parse(sourceData, rowIndex, context);
            if (compressTypeObj != null) {
                compressType = OperatorTools.parseString(compressTypeObj);
            }
        }
        compressType = compressType.toLowerCase();

        try {
            byte[] lengthBytes = intToLowByteArray(str.length());
            CompressHandler handler = CompressionAlgorithmFactory.getCompressHandlerByName(compressType);
            if (handler == null) {
                throw new RuntimeException(compressType + " is not supported.");
            }
            byte[] compressBytes = handler.compress(str.getBytes(CHARSET));
            return mergeByteArray(lengthBytes, compressBytes);
        } catch (Exception e) {
            log.error("Compression failed", e);
            return null;
        }
    }
CompressFunction  ->  compress(string_to_compress[,compress_type(default:deflater)])
description:
- Return NULL if 'string_to_compress' is NULL
- Return "" if 'string_to_compress' is ""
- Return the result as a binary string
Note: This function supports three compression algorithms: deflater, gzip and zip. 'compress_type' defaults to defer.
In addition, in order to output the compressed results in the form of strings, this method uses the ISO_8859_1
character set.
inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/function/compression/CompressFunction.java