@Override
    public List<LogicalSlot> peakSlotsToAllocate(SlotTracker slotTracker) {
        updateOptionsPeriodically();

        List<LogicalSlot> slotsToAllocate = Lists.newArrayList();

        int curNumAllocatedSmallSlots = numAllocatedSmallSlots;
        for (SlotContext slotContext : requiringSmallSlots.values()) {
            LogicalSlot slot = slotContext.getSlot();
            if (curNumAllocatedSmallSlots + slot.getNumPhysicalSlots() > opts.v2().getTotalSmallSlots()) {
                break;
            }

            requiringQueue.remove(slotContext);

            slotsToAllocate.add(slot);
            slotContext.setAllocateAsSmallSlot();
            curNumAllocatedSmallSlots += slot.getNumPhysicalSlots();
        }

        int numAllocatedSlots = slotTracker.getNumAllocatedSlots() - numAllocatedSmallSlots;
        while (!requiringQueue.isEmpty()) {
            SlotContext slotContext = requiringQueue.peak();
            if (!isGlobalSlotAvailable(numAllocatedSlots, slotContext.getSlot())) {
                break;
            }

            requiringQueue.poll();

            slotsToAllocate.add(slotContext.getSlot());
            numAllocatedSlots += slotContext.getSlot().getNumPhysicalSlots();
        }

        return slotsToAllocate;
    }
@Test
    public void testSmallSlot() {
        QueryQueueOptions opts = QueryQueueOptions.createFromEnv();
        SlotSelectionStrategyV2 strategy = new SlotSelectionStrategyV2();
        SlotTracker slotTracker = new SlotTracker(ImmutableList.of(strategy));

        LogicalSlot largeSlot = generateSlot(opts.v2().getTotalSlots() - 1);
        List<LogicalSlot> smallSlots = IntStream.range(0, NUM_CORES + 2)
                .mapToObj(i -> generateSlot(1))
                .collect(Collectors.toList());

        // 1. Require and allocate the large slot with `totalSlots - 1` slots.
        slotTracker.requireSlot(largeSlot);
        List<LogicalSlot> peakSlots = strategy.peakSlotsToAllocate(slotTracker);
        assertThat(peakSlots).containsExactly(largeSlot);
        slotTracker.allocateSlot(largeSlot);

        // 2. Require `NUM_CORES + 2` small slots.
        for (LogicalSlot smallSlot : smallSlots) {
            slotTracker.requireSlot(smallSlot);
        }

        // 3. Could peak and allocate `NUM_CORES + 1` small slots:
        // - `NUM_CORES` small slot are allocated as small slots.
        // - 1 small slot is allocated as a non-small slot.
        List<LogicalSlot> peakSmallSlots = strategy.peakSlotsToAllocate(slotTracker);
        assertThat(peakSmallSlots).hasSize(NUM_CORES + 1);
        for (LogicalSlot peakSmallSlot : peakSmallSlots) {
            slotTracker.allocateSlot(peakSmallSlot);
        }

        // 4. Release the large slot and then the rest one small slot could be peaked.
        peakSlots = strategy.peakSlotsToAllocate(slotTracker);
        assertThat(peakSlots).isEmpty();

        slotTracker.releaseSlot(largeSlot.getSlotId());
        peakSlots = strategy.peakSlotsToAllocate(slotTracker);
        assertThat(peakSlots).hasSize(1);
        slotTracker.allocateSlot(peakSlots.get(0));
        assertThat(slotTracker.getNumAllocatedSlots()).isEqualTo(smallSlots.size());

        // 5. Require `10+numAvailableSlots` small slots.
        final int numAvailableSlots = opts.v2().getTotalSlots() + opts.v2().getTotalSmallSlots() - smallSlots.size();
        List<LogicalSlot> smallSlots2 =
                IntStream.range(0, 10 + numAvailableSlots)
                        .mapToObj(i -> generateSlot(1))
                        .collect(Collectors.toList());
        smallSlots2.forEach(slotTracker::requireSlot);

        // 6. Could peak and allocate `numAvailableSlots` small slots.
        peakSmallSlots = strategy.peakSlotsToAllocate(slotTracker);
        peakSmallSlots.forEach(slotTracker::allocateSlot);
        assertThat(peakSmallSlots).hasSize(numAvailableSlots);
        assertThat(slotTracker.getNumAllocatedSlots()).isEqualTo(opts.v2().getTotalSlots() + opts.v2().getTotalSmallSlots());

        // 7. Release 10 small slots and then the rest small slots could be peaked and allocated.
        smallSlots.stream().limit(10).forEach(slot -> slotTracker.releaseSlot(slot.getSlotId()));
        peakSmallSlots = strategy.peakSlotsToAllocate(slotTracker);
        peakSmallSlots.forEach(slotTracker::allocateSlot);
        assertThat(peakSmallSlots).hasSize(10);
        assertThat(slotTracker.getNumAllocatedSlots()).isEqualTo(opts.v2().getTotalSlots() + opts.v2().getTotalSmallSlots());
    }