presto-main/src/main/java/io/prestosql/operator/LookupJoinOperator.java/LookupJoinOperator/tryUnspillNext
class LookupJoinOperator:        private void tryUnspillNext()
        {
            verify(probe == null);

            if (!partitionedConsumption.isDone()) {
                return;
            }

            if (lookupPartitions == null) {
                lookupPartitions = getDone(partitionedConsumption).beginConsumption();
            }

            if (unspilledInputPages.hasNext()) {
                addInput(unspilledInputPages.next());
                return;
            }

            if (unspilledLookupSource.isPresent()) {
                if (!unspilledLookupSource.get().isDone()) {
                    // Not unspilled yet
                    return;
                }
                LookupSource lookupSource = getDone(unspilledLookupSource.get()).get();
                unspilledLookupSource = Optional.empty();

                // Close previous lookupSourceProvider (either supplied initially or for the previous partition)
                lookupSourceProvider.close();
                lookupSourceProvider = new StaticLookupSourceProvider(lookupSource);
                // If the partition was spilled during processing, its position count will be considered twice.
                statisticsCounter.updateLookupSourcePositions(lookupSource.getJoinPositionCount());

                int partition = currentPartition.get().number();
                unspilledInputPages = spiller.map(spiller -> spiller.getSpilledPages(partition))
                        .orElse(emptyIterator());

                Optional.ofNullable(savedRows.remove(partition)).ifPresent(savedRow -> {
                    restoreProbe(
                            savedRow.row,
                            savedRow.joinPositionWithinPartition,
                            savedRow.currentProbePositionProducedRow,
                            savedRow.joinSourcePositions,
                            SpillInfoSnapshot.noSpill());
                });

                return;
            }

            if (lookupPartitions.hasNext()) {
                currentPartition.ifPresent(Partition::release);
                currentPartition = Optional.of(lookupPartitions.next());
                unspilledLookupSource = Optional.of(currentPartition.get().load());

                return;
            }

            currentPartition.ifPresent(Partition::release);
            if (lookupSourceProvider != null) {
                // There are no more partitions to process, so clean up everything
                lookupSourceProvider.close();
                lookupSourceProvider = null;
            }
            spiller.ifPresent(PartitioningSpiller::verifyAllPartitionsRead);
            finished = true;
        }