presto-main/src/main/java/io/prestosql/operator/index/UnloadedIndexKeyRecordSet.java/UnloadedIndexKeyRecordSet/UnloadedIndexKeyRecordSet
class UnloadedIndexKeyRecordSet:    public UnloadedIndexKeyRecordSet(
            Session session,
            IndexSnapshot existingSnapshot,
            Set<Integer> channelsForDistinct,
            List<Type> types,
            List<UpdateRequest> requests,
            JoinCompiler joinCompiler)
    {
        requireNonNull(existingSnapshot, "existingSnapshot is null");
        this.types = ImmutableList.copyOf(requireNonNull(types, "types is null"));
        requireNonNull(requests, "requests is null");

        int[] distinctChannels = Ints.toArray(channelsForDistinct);
        int[] normalizedDistinctChannels = new int[distinctChannels.length];
        List<Type> distinctChannelTypes = new ArrayList<>(distinctChannels.length);
        for (int i = 0; i < distinctChannels.length; i++) {
            normalizedDistinctChannels[i] = i;
            distinctChannelTypes.add(types.get(distinctChannels[i]));
        }

        ImmutableList.Builder<PageAndPositions> builder = ImmutableList.builder();
        GroupByHash groupByHash = createGroupByHash(session, distinctChannelTypes, normalizedDistinctChannels, Optional.empty(), 10_000, joinCompiler);
        for (UpdateRequest request : requests) {
            Page page = request.getPage();

            Block[] distinctBlocks = new Block[distinctChannels.length];
            for (int i = 0; i < distinctBlocks.length; i++) {
                distinctBlocks[i] = page.getBlock(distinctChannels[i]);
            }

            // Move through the positions while advancing the cursors in lockstep
            Work<GroupByIdBlock> work = groupByHash.getGroupIds(new Page(distinctBlocks));
            boolean done = work.process();
            // TODO: this class does not yield wrt memory limit; enable it
            verify(done);
            GroupByIdBlock groupIds = work.getResult();
            int positionCount = page.getBlock(0).getPositionCount();
            long nextDistinctId = -1;
            checkArgument(groupIds.getGroupCount() <= Integer.MAX_VALUE);
            IntList positions = new IntArrayList((int) groupIds.getGroupCount());
            for (int position = 0; position < positionCount; position++) {
                // We are reading ahead in the cursors, so we need to filter any nulls since they can not join
                if (!containsNullValue(position, page)) {
                    // Only include the key if it is not already in the index
                    if (existingSnapshot.getJoinPosition(position, page) == UNLOADED_INDEX_KEY) {
                        // Only add the position if we have not seen this tuple before (based on the distinct channels)
                        long groupId = groupIds.getGroupId(position);
                        if (nextDistinctId < groupId) {
                            nextDistinctId = groupId;
                            positions.add(position);
                        }
                    }
                }
            }

            if (!positions.isEmpty()) {
                builder.add(new PageAndPositions(request, positions));
            }
        }

        pageAndPositions = builder.build();
    }