presto-main/src/main/java/io/prestosql/operator/index/IndexSnapshotBuilder.java/IndexSnapshotBuilder/createIndexSnapshot
class IndexSnapshotBuilder:    public IndexSnapshot createIndexSnapshot(UnloadedIndexKeyRecordSet indexKeysRecordSet)
    {
        checkArgument(indexKeysRecordSet.getColumnTypes().equals(missingKeysTypes), "indexKeysRecordSet must have same schema as missingKeys");
        checkState(!isMemoryExceeded(), "Max memory exceeded");
        for (Page page : pages) {
            outputPagesIndex.addPage(page);
        }
        pages.clear();

        LookupSource lookupSource = outputPagesIndex.createLookupSourceSupplier(session, keyOutputChannels, keyOutputHashChannel, Optional.empty(), Optional.empty(), ImmutableList.of()).get();

        // Build a page containing the keys that produced no output rows, so in future requests can skip these keys
        verify(missingKeysPageBuilder.isEmpty());
        UnloadedIndexKeyRecordCursor indexKeysRecordCursor = indexKeysRecordSet.cursor();
        while (indexKeysRecordCursor.advanceNextPosition()) {
            Page page = indexKeysRecordCursor.getPage();
            int position = indexKeysRecordCursor.getPosition();
            if (lookupSource.getJoinPosition(position, page, page) < 0) {
                missingKeysPageBuilder.declarePosition();
                for (int i = 0; i < page.getChannelCount(); i++) {
                    Block block = page.getBlock(i);
                    Type type = indexKeysRecordCursor.getType(i);
                    type.appendTo(block, position, missingKeysPageBuilder.getBlockBuilder(i));
                }
            }
        }
        Page missingKeysPage = missingKeysPageBuilder.build();
        if (!missingKeysPageBuilder.isEmpty()) {
            missingKeysPageBuilder.reset();
        }

        memoryInBytes += missingKeysPage.getSizeInBytes();
        if (isMemoryExceeded()) {
            return null;
        }

        // only update missing keys if we have new missing keys
        if (missingKeysPage.getPositionCount() != 0) {
            missingKeysIndex.addPage(missingKeysPage);
            missingKeys = missingKeysIndex.createLookupSourceSupplier(session, missingKeysChannels).get();
        }

        return new IndexSnapshot(lookupSource, missingKeys);
    }