new Range implementation

src/main/java/com/facebook/presto/AggregationUtil.java

@@ -1,19 +1,19 @@
 package com.facebook.presto;
 
 import com.google.common.base.Preconditions;
-import com.google.common.collect.Range;
+
 
 public class AggregationUtil {
-    public static void processGroup(SeekableIterator<ValueBlock> aggregationSource, AggregationFunction aggregation, Range<Long> positions)
+    public static void processGroup(SeekableIterator<ValueBlock> aggregationSource, AggregationFunction aggregation, Range positions)
     {
         RangePositionBlock positionBlock = new RangePositionBlock(positions);
 
         // goto start of range
-        aggregationSource.seekTo(positions.lowerEndpoint());
+        aggregationSource.seekTo(positions.getStart());
         Preconditions.checkState(aggregationSource.hasNext(), "Group start position not found in aggregation source");
 
         // while we have data...
-        while (aggregationSource.hasNext() && aggregationSource.peek().getRange().isConnected(positions)) {
+        while (aggregationSource.hasNext() && aggregationSource.peek().getRange().overlaps(positions)) {
             // process aggregation
             aggregation.add(aggregationSource.next(), positionBlock);
         }

src/main/java/com/facebook/presto/Block.java

@@ -1,6 +1,6 @@
 package com.facebook.presto;
 
-import com.google.common.collect.Range;
+
 
 public interface Block
 {
@@ -13,5 +13,5 @@ public interface Block
 
     Iterable<Long> getPositions();
 
-    Range<Long> getRange();
+    Range getRange();
 }

src/main/java/com/facebook/presto/BlockBuilder.java

@@ -1,7 +1,7 @@
 package com.facebook.presto;
 
 import com.google.common.base.Preconditions;
-import com.google.common.collect.Ranges;
+
 import io.airlift.units.DataSize;
 import io.airlift.units.DataSize.Unit;
 
@@ -87,6 +87,6 @@ public ValueBlock build()
             return EmptyValueBlock.INSTANCE;
         }
 
-        return new UncompressedValueBlock(Ranges.closed(startPosition, startPosition + count - 1), tupleInfo, sliceOutput.slice());
+        return new UncompressedValueBlock(Range.create(startPosition, startPosition + count - 1), tupleInfo, sliceOutput.slice());
     }
 }

src/main/java/com/facebook/presto/DataScan3.java

@@ -23,7 +23,7 @@ public DataScan3(Iterator<? extends ValueBlock> source, Iterator<? extends Posit
     protected ValueBlock computeNext()
     {
         while (advance()) {
-            if (valueBlock.getRange().isConnected(positionBlock.getRange())) {
+            if (valueBlock.getRange().overlaps(positionBlock.getRange())) {
                 ValueBlock result = valueBlock.filter(positionBlock);
                 if (!result.isEmpty()) {
                     return result;
@@ -38,7 +38,7 @@ protected ValueBlock computeNext()
     private boolean advance()
     {
         if (valueBlock != null && positionBlock != null) {
-            if (valueBlock.getRange().upperEndpoint() < positionBlock.getRange().upperEndpoint()) {
+            if (valueBlock.getRange().getEnd() < positionBlock.getRange().getEnd()) {
                 valueBlock = null;
             }
             else {

src/main/java/com/facebook/presto/EmptyPositionBlock.java

@@ -1,8 +1,8 @@
 package com.facebook.presto;
 
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Range;
-import com.google.common.collect.Ranges;
+
+
 
 import javax.annotation.Nullable;
 
@@ -56,9 +56,9 @@ public Iterable<Long> getPositions()
     }
 
     @Override
-    public Range<Long> getRange()
+    public Range getRange()
     {
-        return Ranges.open(0L, 0L);
+        throw new UnsupportedOperationException("Empty block doesn't have a range");
     }
 
     @Override

src/main/java/com/facebook/presto/EmptyValueBlock.java

@@ -4,8 +4,8 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterators;
 import com.google.common.collect.PeekingIterator;
-import com.google.common.collect.Range;
-import com.google.common.collect.Ranges;
+
+
 
 import java.util.Iterator;
 
@@ -85,9 +85,9 @@ public Iterable<Long> getPositions()
     }
 
     @Override
-    public Range<Long> getRange()
+    public Range getRange()
     {
-        return Ranges.open(0L, 0L);
+        throw new UnsupportedOperationException("Empty block doesn't have a range");
     }
 
     @Override

src/main/java/com/facebook/presto/ForwardingSeekableIterator.java

@@ -55,7 +55,7 @@ public boolean hasNext()
     public boolean seekTo(long position)
     {
         if (last != null) {
-            Preconditions.checkArgument(position >= last.getRange().lowerEndpoint(), "Cannot seek to position %s before current range [%s, %s]", position, last.getRange().lowerEndpoint(), last.getRange().upperEndpoint());
+            Preconditions.checkArgument(position >= last.getRange().getStart(), "Cannot seek to position %s before current range [%s, %s]", position, last.getRange().getStart(), last.getRange().getEnd());
             if (last.getRange().contains(position)) {
                 seekedTo = last;
                 return true;
@@ -69,7 +69,7 @@ public boolean seekTo(long position)
                 seekedTo = last;
                 return true;
             }
-            else if (last.getRange().lowerEndpoint() > position) {
+            else if (last.getRange().getStart() > position) {
                 seekedTo = last;
                 return false;
             }

src/main/java/com/facebook/presto/GroupBy.java

@@ -2,8 +2,8 @@
 
 import com.google.common.collect.AbstractIterator;
 import com.google.common.collect.PeekingIterator;
-import com.google.common.collect.Range;
-import com.google.common.collect.Ranges;
+
+
 
 import java.util.Iterator;
 
@@ -57,7 +57,7 @@ protected RunLengthEncodedBlock computeNext()
         }
 
         long endPosition = entry.getPosition();
-        Range<Long> range = Ranges.closed(startPosition, endPosition);
+        Range range = Range.create(startPosition, endPosition);
 
         RunLengthEncodedBlock group = new RunLengthEncodedBlock(groupByKey, range);
         return group;

src/main/java/com/facebook/presto/MaskedValueBlock.java

@@ -6,7 +6,6 @@
 import com.google.common.base.Predicate;
 import com.google.common.collect.Iterators;
 import com.google.common.collect.PeekingIterator;
-import com.google.common.collect.Range;
 
 import java.util.Iterator;
 
@@ -21,11 +20,12 @@ public static ValueBlock maskBlock(ValueBlock valueBlock, PositionBlock position
             return EmptyValueBlock.INSTANCE;
         }
 
-        Range<Long> intersection = valueBlock.getRange().intersection(positions.getRange());
-        if (intersection.isEmpty()) {
+        if (!valueBlock.getRange().overlaps(positions.getRange())) {
             return EmptyValueBlock.INSTANCE;
         }
 
+        Range intersection = valueBlock.getRange().intersect(positions.getRange());
+
         if (valueBlock.isSingleValue() && positions.isPositionsContiguous()) {
             Tuple value = valueBlock.iterator().next();
             return new RunLengthEncodedBlock(value, intersection);
@@ -133,7 +133,7 @@ public Iterable<Long> getPositions()
     }
 
     @Override
-    public Range<Long> getRange()
+    public Range getRange()
     {
         return positionBlock.getRange();
     }

src/main/java/com/facebook/presto/PackedLongSerde.java

@@ -2,21 +2,21 @@
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.AbstractIterator;
-import com.google.common.collect.Range;
-import com.google.common.collect.Ranges;
+
+
 
 import java.util.Iterator;
 
 public class PackedLongSerde
 {
     private final byte bitWidth;
-    private final Range<Long> allowedRange;
+    private final Range allowedRange;
 
     public PackedLongSerde(int bitWidth)
     {
         Preconditions.checkArgument(bitWidth > 0 && bitWidth <= Long.SIZE);
         this.bitWidth = (byte) bitWidth;
-        this.allowedRange = Ranges.closed(-1L << (bitWidth - 1), ~(-1L << (bitWidth - 1)));
+        this.allowedRange = Range.create(-1L << (bitWidth - 1), ~(-1L << (bitWidth - 1)));
     }
 
     public void serialize(Iterable<Long> items, SliceOutput sliceOutput)

src/main/java/com/facebook/presto/Range.java

@@ -0,0 +1,115 @@
+package com.facebook.presto;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.AbstractIterator;
+
+import java.util.Iterator;
+
+public class Range
+    implements Iterable<Long>
+{
+    private final long start;
+    private final long end;
+
+    public Range(long start, long end)
+    {
+        Preconditions.checkArgument(start <= end, "start (%s) must be <= end (%s)", start, end);
+
+        this.start = start;
+        this.end = end;
+    }
+
+    public static Range create(long start, long end)
+    {
+        return new Range(start, end);
+    }
+
+    public long getStart()
+    {
+        return start;
+    }
+
+    public long getEnd()
+    {
+        return end;
+    }
+
+    public long length()
+    {
+        return end - start + 1;
+    }
+
+    public boolean contains(long value)
+    {
+        return value >= start && value <= end;
+    }
+
+    public boolean overlaps(Range other)
+    {
+        return start <= other.end && other.start <= end;
+    }
+
+    public Range intersect(Range other)
+    {
+        Preconditions.checkArgument(overlaps(other), "Ranges do not overlap %s vs %s", this, other);
+
+        return create(Math.max(start, other.start), Math.min(end, other.end));
+    }
+
+    public String toString()
+    {
+        return String.format("[%s..%s]", start, end);
+    }
+
+    @Override
+    public boolean equals(Object o)
+    {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        Range range = (Range) o;
+
+        if (end != range.end) {
+            return false;
+        }
+        if (start != range.start) {
+            return false;
+        }
+
+        return true;
+    }
+
+    @Override
+    public int hashCode()
+    {
+        int result = (int) (start ^ (start >>> 32));
+        result = 31 * result + (int) (end ^ (end >>> 32));
+        return result;
+    }
+
+    @Override
+    public Iterator<Long> iterator()
+    {
+        return new AbstractIterator<Long>()
+        {
+            private long current = start;
+
+            @Override
+            protected Long computeNext()
+            {
+                if (current > end) {
+                    endOfData();
+                    return null;
+                }
+
+                long result = current;
+                ++current;
+                return result;
+            }
+        };
+    }
+}