Get rid of empty blocks

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

@@ -4,7 +4,6 @@
 
 public interface Block
 {
-    boolean isEmpty();
     int getCount();
 
     boolean isSorted();

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

@@ -1,7 +1,6 @@
 package com.facebook.presto;
 
 import com.google.common.base.Preconditions;
-
 import io.airlift.units.DataSize;
 import io.airlift.units.DataSize.Unit;
 
@@ -83,9 +82,7 @@ public ValueBlock build()
     {
         flushTupleIfNecessary();
 
-        if (count == 0) {
-            return EmptyValueBlock.INSTANCE;
-        }
+        Preconditions.checkState(count > 0, "Cannot build an empty block");
 
         return new UncompressedValueBlock(Range.create(startPosition, startPosition + count - 1), tupleInfo, sliceOutput.slice());
     }

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

@@ -1,5 +1,6 @@
 package com.facebook.presto;
 
+import com.google.common.base.Optional;
 import com.google.common.base.Predicate;
 import com.google.common.collect.AbstractIterator;
 
@@ -21,9 +22,9 @@ public DataScan1(Iterator<ValueBlock> source, Predicate<Tuple> predicate)
     protected PositionBlock computeNext()
     {
         while (source.hasNext()) {
-            PositionBlock block = source.next().selectPositions(predicate);
-            if (!block.isEmpty()) {
-                return block;
+            Optional<PositionBlock> block = source.next().selectPositions(predicate);
+            if (block.isPresent()) {
+                return block.get();
             }
         }
 

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

@@ -1,5 +1,6 @@
 package com.facebook.presto;
 
+import com.google.common.base.Optional;
 import com.google.common.base.Predicate;
 import com.google.common.collect.AbstractIterator;
 
@@ -21,9 +22,9 @@ public DataScan2(Iterator<ValueBlock> source, Predicate<Tuple> predicate)
     protected ValueBlock computeNext()
     {
         while (source.hasNext()) {
-            ValueBlock block = source.next().selectPairs(predicate);
-            if (!block.isEmpty()) {
-                return block;
+            Optional<ValueBlock> block = source.next().selectPairs(predicate);
+            if (block.isPresent()) {
+                return block.get();
             }
         }
 

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

@@ -1,5 +1,6 @@
 package com.facebook.presto;
 
+import com.google.common.base.Optional;
 import com.google.common.collect.AbstractIterator;
 
 import java.util.Iterator;
@@ -24,9 +25,9 @@ protected ValueBlock computeNext()
     {
         while (advance()) {
             if (valueBlock.getRange().overlaps(positionBlock.getRange())) {
-                ValueBlock result = valueBlock.filter(positionBlock);
-                if (!result.isEmpty()) {
-                    return result;
+                Optional<ValueBlock> result = valueBlock.filter(positionBlock);
+                if (result.isPresent()) {
+                    return result.get();
                 }
             }
         }

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

@@ -3,6 +3,7 @@
  */
 package com.facebook.presto;
 
+import com.google.common.base.Optional;
 import com.google.common.base.Predicate;
 import com.google.common.collect.Iterators;
 import com.google.common.collect.PeekingIterator;
@@ -14,32 +15,32 @@
 
 public class MaskedValueBlock implements ValueBlock
 {
-    public static ValueBlock maskBlock(ValueBlock valueBlock, PositionBlock positions)
+    public static Optional<ValueBlock> maskBlock(ValueBlock valueBlock, PositionBlock positions)
     {
-        if (positions.isEmpty()) {
-            return EmptyValueBlock.INSTANCE;
-        }
-
         if (!valueBlock.getRange().overlaps(positions.getRange())) {
-            return EmptyValueBlock.INSTANCE;
+            return Optional.absent();
         }
 
         Range intersection = valueBlock.getRange().intersect(positions.getRange());
 
         if (valueBlock.isSingleValue() && positions.isPositionsContiguous()) {
             Tuple value = valueBlock.iterator().next();
-            return new RunLengthEncodedBlock(value, intersection);
+            return Optional.<ValueBlock>of(new RunLengthEncodedBlock(value, intersection));
         }
 
-        PositionBlock newPositionBlock;
+        Optional<PositionBlock> newPositionBlock;
         if (valueBlock.isPositionsContiguous()) {
             newPositionBlock = positions.filter(new RangePositionBlock(valueBlock.getRange()));
         }
         else {
             newPositionBlock = positions.filter(valueBlock.toPositionBlock());
         }
 
-        return new MaskedValueBlock(valueBlock, newPositionBlock);
+        if (newPositionBlock.isPresent()) {
+            return Optional.<ValueBlock>of(new MaskedValueBlock(valueBlock, newPositionBlock.get()));
+        }
+
+        return Optional.absent();
     }
 
     private final ValueBlock valueBlock;
@@ -52,13 +53,13 @@ private MaskedValueBlock(ValueBlock valueBlock, PositionBlock positionBlock)
     }
 
     @Override
-    public PositionBlock selectPositions(Predicate<Tuple> predicate)
+    public Optional<PositionBlock> selectPositions(Predicate<Tuple> predicate)
     {
         throw new UnsupportedOperationException();
     }
 
     @Override
-    public ValueBlock selectPairs(Predicate<Tuple> predicate)
+    public Optional<ValueBlock> selectPairs(Predicate<Tuple> predicate)
     {
         throw new UnsupportedOperationException();
     }
@@ -70,17 +71,13 @@ public PositionBlock toPositionBlock()
     }
 
     @Override
-    public ValueBlock filter(PositionBlock positions)
+    public Optional<ValueBlock> filter(PositionBlock positions)
     {
-        if (positions.isEmpty()) {
-            return EmptyValueBlock.INSTANCE;
-        }
-
-        PositionBlock activePositions = positionBlock.filter(positions);
-        if (activePositions.isEmpty()) {
-            return EmptyValueBlock.INSTANCE;
+        Optional<PositionBlock> activePositions = positionBlock.filter(positions);
+        if (!activePositions.isPresent()) {
+            return Optional.absent();
         }
-        return new MaskedValueBlock(valueBlock, activePositions);
+        return Optional.<ValueBlock>of(new MaskedValueBlock(valueBlock, activePositions.get()));
     }
 
     @Override
@@ -96,12 +93,6 @@ public boolean apply(Pair pair)
         }));
     }
 
-    @Override
-    public boolean isEmpty()
-    {
-        return false;
-    }
-
     @Override
     public int getCount()
     {

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

@@ -1,9 +1,10 @@
 package com.facebook.presto;
 
+import com.google.common.base.Optional;
 import com.google.common.base.Predicate;
 
 public interface PositionBlock
     extends Block, Predicate<Long>
 {
-    PositionBlock filter(PositionBlock positionBlock);
+    Optional<PositionBlock> filter(PositionBlock positionBlock);
 }