### Question:
GenericPartitioningSpiller implements PartitioningSpiller { @Override public synchronized PartitioningSpillResult partitionAndSpill(Page page, IntPredicate spillPartitionMask) { requireNonNull(page, "page is null"); requireNonNull(spillPartitionMask, "spillPartitionMask is null"); checkArgument(page.getChannelCount() == types.size(), "Wrong page channel count, expected %s but got %s", types.size(), page.getChannelCount()); checkState(!readingStarted, "reading already started"); IntArrayList unspilledPositions = partitionPage(page, spillPartitionMask); ListenableFuture<?> future = flushFullBuilders(); return new PartitioningSpillResult(future, page.getPositions(unspilledPositions.elements(), 0, unspilledPositions.size())); } GenericPartitioningSpiller(
            List<Type> types,
            PartitionFunction partitionFunction,
            SpillContext spillContext,
            AggregatedMemoryContext memoryContext,
            SingleStreamSpillerFactory spillerFactory); @Override synchronized Iterator<Page> getSpilledPages(int partition); @Override synchronized void verifyAllPartitionsRead(); @Override synchronized PartitioningSpillResult partitionAndSpill(Page page, IntPredicate spillPartitionMask); @Override synchronized void close();  }### Answer:
@Test public void testFileSpiller() throws Exception { try (PartitioningSpiller spiller = factory.create( TYPES, new FourFixedPartitionsPartitionFunction(0), mockSpillContext(), mockMemoryContext(scheduledExecutor))) { RowPagesBuilder builder = RowPagesBuilder.rowPagesBuilder(TYPES); builder.addSequencePage(10, SECOND_PARTITION_START, 5, 10, 15); builder.addSequencePage(10, FIRST_PARTITION_START, -5, 0, 5); List<Page> firstSpill = builder.build(); builder = RowPagesBuilder.rowPagesBuilder(TYPES); builder.addSequencePage(10, THIRD_PARTITION_START, 15, 20, 25); builder.addSequencePage(10, FOURTH_PARTITION_START, 25, 30, 35); List<Page> secondSpill = builder.build(); IntPredicate spillPartitionMask = ImmutableSet.of(1, 2)::contains; PartitioningSpillResult result = spiller.partitionAndSpill(firstSpill.get(0), spillPartitionMask); result.getSpillingFuture().get(); assertEquals(result.getRetained().getPositionCount(), 0); result = spiller.partitionAndSpill(firstSpill.get(1), spillPartitionMask); result.getSpillingFuture().get(); assertEquals(result.getRetained().getPositionCount(), 10); result = spiller.partitionAndSpill(secondSpill.get(0), spillPartitionMask); result.getSpillingFuture().get(); assertEquals(result.getRetained().getPositionCount(), 0); result = spiller.partitionAndSpill(secondSpill.get(1), spillPartitionMask); result.getSpillingFuture().get(); assertEquals(result.getRetained().getPositionCount(), 10); builder = RowPagesBuilder.rowPagesBuilder(TYPES); builder.addSequencePage(10, SECOND_PARTITION_START, 5, 10, 15); List<Page> secondPartition = builder.build(); builder = RowPagesBuilder.rowPagesBuilder(TYPES); builder.addSequencePage(10, THIRD_PARTITION_START, 15, 20, 25); List<Page> thirdPartition = builder.build(); assertSpilledPages( TYPES, spiller, ImmutableList.of(ImmutableList.of(), secondPartition, thirdPartition, ImmutableList.of())); } }