@@ -0,0 +1,2323 @@
+package com.facebook.presto.operators;%0A%0Aimport com.facebook.presto.BlockStream;%0Aimport com.facebook.presto.Cursor;%0Aimport com.facebook.presto.Range;%0Aimport com.facebook.presto.RunLengthEncodedBlock;%0Aimport com.facebook.presto.RunLengthEncodedCursor;%0Aimport com.facebook.presto.Slice;%0Aimport com.facebook.presto.Tuple;%0Aimport com.facebook.presto.TupleInfo;%0Aimport com.facebook.presto.ValueBlock;%0Aimport com.google.common.collect.AbstractIterator;%0A%0Aimport java.util.Iterator;%0A%0A/**%0A * Group input data and produce a single block for each sequence of identical values.%0A */%0Apublic class GroupByBlockStream%0A    implements BlockStream%3CRunLengthEncodedBlock%3E%0A%7B%0A    private final BlockStream%3C? extends ValueBlock%3E source;%0A%0A    public GroupByBlockStream(BlockStream%3C? extends ValueBlock%3E keySource)%0A    %7B%0A        this.source = keySource;%0A    %7D%0A%0A    @Override%0A    public TupleInfo getTupleInfo()%0A    %7B%0A        return source.getTupleInfo();%0A    %7D%0A%0A    @Override%0A    public Cursor cursor()%0A    %7B%0A        return new RunLengthEncodedCursor(getTupleInfo(), iterator());%0A    %7D%0A%0A    @Override%0A    public Iterator%3CRunLengthEncodedBlock%3E iterator()%0A    %7B%0A        final Cursor cursor = source.cursor();%0A        cursor.advanceNextValue();%0A%0A        return new AbstractIterator%3CRunLengthEncodedBlock%3E()%0A        %7B%0A            @Override%0A            protected RunLengthEncodedBlock computeNext()%0A            %7B%0A                // if no more data, return null%0A                if (!cursor.hasNextValue()) %7B%0A                    endOfData();%0A                    return null;%0A                %7D%0A%0A                Slice key = cursor.getSlice(0);%0A%0A                long startPosition = cursor.getPosition();%0A%0A                if (!cursor.hasNextValue()) %7B%0A                    Range range = Range.create(startPosition, startPosition);%0A                    return new RunLengthEncodedBlock(new Tuple(key, getTupleInfo()), range);%0A                %7D%0A%0A                do %7B%0A                    cursor.advanceNextValue();%0A                %7D%0A                while (cursor.equals(0, key) && cursor.hasNextValue());%0A%0A%0A                long endPosition = cursor.getPosition() - 1;%0A                Range range = Range.create(startPosition, endPosition);%0A%0A                return new RunLengthEncodedBlock(new Tuple(key, getTupleInfo()), range);%0A            %7D%0A        %7D;%0A    %7D%0A%7D%0A

Implement cursor-based group by