@Override
    @NotNull
    public List<PartitionStatistics> select(Collection<PartitionStatistics> statistics, Set<Long> excludeTables) {
        long now = System.currentTimeMillis();
        return statistics.stream()
                .filter(p -> p.getNextCompactionTime() <= now)
                .filter(p -> !excludeTables.contains(p.getPartition().getTableId()))
                .filter(p -> isReadyForCompaction(p, now))
                .collect(Collectors.toList());
    }
@Test
    public void testCompactionTimeNotReached() {
        List<PartitionStatistics> statisticsList = new ArrayList<>();

        final PartitionIdentifier partitionIdentifier = new PartitionIdentifier(1, 2, 4);
        PartitionStatistics statistics = new PartitionStatistics(partitionIdentifier);
        statistics.setCompactionVersion(new PartitionVersion(1, 0));
        statistics.setCurrentVersion(new PartitionVersion(MIN_COMPACTION_VERSIONS + 1, System.currentTimeMillis()));
        statisticsList.add(statistics);

        statistics.setNextCompactionTime(System.currentTimeMillis() + 60 * 1000);

        Assert.assertEquals(0, selector.select(statisticsList, new HashSet<Long>()).size());

        statistics.setNextCompactionTime(System.currentTimeMillis() - 10);

        Assert.assertSame(statistics, selector.select(statisticsList, new HashSet<Long>()).get(0));
    }