presto-main/src/main/java/io/prestosql/operator/LookupJoinOperator.java/LookupJoinOperator/processProbe
class LookupJoinOperator:        private void processProbe()
        {
            verifyNotNull(probe);

            Optional<SpillInfoSnapshot> spillInfoSnapshotIfSpillChanged = lookupSourceProvider.withLease(lookupSourceLease -> {
                if (lookupSourceLease.spillEpoch() == inputPageSpillEpoch) {
                    // Spill state didn't change, so process as usual.
                    processProbe(lookupSourceLease.getLookupSource());
                    return Optional.empty();
                }

                return Optional.of(SpillInfoSnapshot.from(lookupSourceLease));
            });

            if (!spillInfoSnapshotIfSpillChanged.isPresent()) {
                return;
            }
            SpillInfoSnapshot spillInfoSnapshot = spillInfoSnapshotIfSpillChanged.get();
            long joinPositionWithinPartition;
            if (joinPosition >= 0) {
                joinPositionWithinPartition = lookupSourceProvider.withLease(lookupSourceLease -> lookupSourceLease.getLookupSource().joinPositionWithinPartition(joinPosition));
            }
            else {
                joinPositionWithinPartition = -1;
            }

            /*
             * Spill state changed. All probe rows that were not processed yet should be treated as regular input (and be partially spilled).
             * If current row maps to the now-spilled partition, it needs to be saved for later. If it maps to a partition still in memory, it
             * should be added together with not-yet-processed rows. In either case we need to resume processing the row starting at its
             * current position in the lookup source.
             */
            verify(spillInfoSnapshot.hasSpilled());
            verify(spillInfoSnapshot.getSpillEpoch() > inputPageSpillEpoch);

            Page currentPage = probe.getPage();
            int currentPosition = probe.getPosition();
            long currentJoinPosition = this.joinPosition;
            boolean currentProbePositionProducedRow = this.currentProbePositionProducedRow;

            clearProbe();

            if (currentPosition < 0) {
                // Processing of the page hasn't been started yet.
                addInput(currentPage, spillInfoSnapshot);
            }
            else {
                int currentRowPartition = getPartitionGenerator().getPartition(currentPage, currentPosition);
                boolean currentRowSpilled = spillInfoSnapshot.getSpillMask().test(currentRowPartition);

                if (currentRowSpilled) {
                    savedRows.merge(
                            currentRowPartition,
                            new SavedRow(currentPage, currentPosition, joinPositionWithinPartition, currentProbePositionProducedRow, joinSourcePositions),
                            (oldValue, newValue) -> {
                                throw new IllegalStateException(format("Partition %s is already spilled", currentRowPartition));
                            });
                    joinSourcePositions = 0;
                    Page unprocessed = pageTail(currentPage, currentPosition + 1);
                    addInput(unprocessed, spillInfoSnapshot);
                }
                else {
                    Page remaining = pageTail(currentPage, currentPosition);
                    restoreProbe(remaining, currentJoinPosition, currentProbePositionProducedRow, joinSourcePositions, spillInfoSnapshot);
                }
            }
        }