class S3SelectAccessor extends BasePlugin implements Accessor {

    InputSerialization getInputSerialization(RequestContext context) {
        InputSerialization inputSerialization = new InputSerialization();

        // We need to infer the format name from the profile (i.e. s3:parquet
        // would return parquet for the format)
        String format = context.inferFormatName();
        String compressionType = context.getOption(COMPRESSION_TYPE);

        LOG.debug("With format {}", format);
        if (StringUtils.equalsIgnoreCase(format, "parquet")) {
            inputSerialization.setParquet(new ParquetInput());
        } else if (StringUtils.equalsIgnoreCase(format, "json")) {
            inputSerialization.setJson(getJSONInput(context));
        } else {
            inputSerialization.setCsv(getCSVInput(context));
        }

        LOG.debug("With compression type {}", compressionType);
        if (StringUtils.equalsIgnoreCase(compressionType, "gzip")) {
            inputSerialization.setCompressionType(CompressionType.GZIP);
        } else if (StringUtils.equalsIgnoreCase(compressionType, "bzip2")) {
            inputSerialization.setCompressionType(CompressionType.BZIP2);
        } else {
            inputSerialization.setCompressionType(CompressionType.NONE);
        }

        return inputSerialization;
    }

}