@Override
    @NotNull
    public List<PartitionStatistics> select(@NotNull Collection<PartitionStatistics> statistics,
            @NotNull Set<Long> excludeTables) {
        double minScore = Config.lake_compaction_score_selector_min_score;
        long now = System.currentTimeMillis();
        return statistics.stream()
                .filter(p -> p.getCompactionScore() != null)
                .filter(p -> !excludeTables.contains(p.getPartition().getTableId()))
                // When manual compaction is triggered, we just skip min score and time check
                .filter(p -> (p.getPriority() != PartitionStatistics.CompactionPriority.DEFAULT
                        || (p.getNextCompactionTime() <= now && p.getCompactionScore().getMax() >= minScore)))
                .collect(Collectors.toList());
    }
@Test
    public void test() {
        List<PartitionStatistics> statisticsList = new ArrayList<>();
        PartitionStatistics statistics = new PartitionStatistics(new PartitionIdentifier(1, 2, 3));
        statistics.setCompactionScore(Quantiles.compute(Collections.singleton(0.0)));
        statisticsList.add(statistics);

        statistics = new PartitionStatistics(new PartitionIdentifier(1, 2, 4));
        statistics.setCompactionScore(Quantiles.compute(Collections.singleton(0.99)));
        statisticsList.add(statistics);

        statistics = new PartitionStatistics(new PartitionIdentifier(1, 2, 5));
        statistics.setCompactionScore(Quantiles.compute(Collections.singleton(1.0)));
        statisticsList.add(statistics);

        statistics = new PartitionStatistics(new PartitionIdentifier(1, 2, 6));
        statistics.setCompactionScore(Quantiles.compute(Collections.singleton(1.1)));
        statisticsList.add(statistics);

        List<PartitionStatistics> targetList = selector.select(statisticsList, new HashSet<Long>());
        Assert.assertEquals(2, targetList.size());
        Assert.assertEquals(5, targetList.get(0).getPartition().getPartitionId());
        Assert.assertEquals(6, targetList.get(1).getPartition().getPartitionId());
    }