Percentiles

percentiles.java

    @GetMapping("/percentiles-distribution")
    public Flux<PercentilesDistribution> getAggregatePercentiles(
            @RequestParam @Positive long masterId,
            @RequestParam(defaultValue = "0") @Min(-1) long from,
            @RequestParam(defaultValue = "0") @Min(-1) long to,
            @RequestParam(defaultValue = "ALL") @Size(max = Constants.MAX_LOCATIONS) ArrayList<String> locationId,
            @RequestParam(defaultValue = "") @Size(max = Constants.MAX_SCENARIOS) ArrayList<String> scenarioId,
            @RequestParam(defaultValue = "") ArrayList<String> labelIds
    ) {
        final int interval = 60;
        final String collectionName = collectionService.getCollectionName(interval);
        UUID uuid = UUID.randomUUID();

        log.info("UUID: {}, Querying {} scenarios, {} locations", uuid, scenarioId.size(), locationId.size());

        Query query = getQuery(masterId, from, to, locationId, scenarioId, labelIds);

        StopWatch stopWatch = new StopWatch("performance");
        stopWatch.start("mongo");

        return mongoService
                .find(query, Record.class, collectionName)
                .collectMultimap(Record::getLabelId)
                .doOnNext(records -> {
                    stopWatch.stop();
                    stopWatch.start("calculations");
                })
                .flatMapIterable(Map::entrySet)
                .flatMap(entry -> Flux.fromIterable(entry.getValue())
                        .map(record -> {
                            try {
                                return IntCountsHistogram.decodeFromCompressedByteBuffer(ByteBuffer.wrap(Base64.getDecoder().decode(record.getHstRt())), 0);
                            } catch (DataFormatException e) {
                                log.warn("Could not decode histogram from record", e);
                                return null;
                            }
                        })
                        .filter(Objects::nonNull)
                        .reduce((histograms, histogram) -> {
                            if (histograms == null) {
                                return histogram;
                            }
                            histograms.add(histogram);
                            return histograms;
                        })
                        .flatMap(aggregatedHistogram -> Flux.range(0, 101)
                                .map(aggregatedHistogram::getValueAtPercentile)
                                .map(Double::valueOf)
                                .collectList())
                        .map(dists -> {
                            final PercentilesDistribution record = new PercentilesDistribution();
                            record.setLabelId(entry.getKey());
                            record.distribution.addAll(dists);
                            return record;
                        })
                );
    }

    @Data
    public static class PercentilesDistribution {
        public String labelId;

        public ArrayList<Double> distribution = new ArrayList<>();
    }