Implement count aggregation

src/main/java/com/facebook/presto/aggregations/CountAggregation.java

@@ -0,0 +1,37 @@
+package com.facebook.presto.aggregations;
+
+import com.facebook.presto.PositionBlock;
+import com.facebook.presto.Tuple;
+import com.facebook.presto.TupleInfo;
+import com.facebook.presto.ValueBlock;
+import com.google.common.base.Optional;
+
+public class CountAggregation
+        implements AggregationFunction
+{
+    private long count;
+
+    @Override
+    public TupleInfo getTupleInfo()
+    {
+        return new TupleInfo(TupleInfo.Type.FIXED_INT_64);
+    }
+
+    @Override
+    public void add(ValueBlock values, PositionBlock relevantPositions)
+    {
+        Optional<ValueBlock> filtered = values.filter(relevantPositions);
+
+        if (filtered.isPresent()) {
+            count += filtered.get().getCount();
+        }
+    }
+
+    @Override
+    public Tuple evaluate()
+    {
+        return getTupleInfo().builder()
+                .append(count)
+                .build();
+    }
+}

src/test/java/com/facebook/presto/Blocks.java

@@ -0,0 +1,30 @@
+package com.facebook.presto;
+
+import static com.facebook.presto.TupleInfo.Type.FIXED_INT_64;
+import static com.facebook.presto.TupleInfo.Type.VARIABLE_BINARY;
+import static com.google.common.base.Charsets.UTF_8;
+
+public class Blocks
+{
+    public static ValueBlock createBlock(int position, String... values)
+    {
+        BlockBuilder builder = new BlockBuilder(position, new TupleInfo(VARIABLE_BINARY));
+
+        for (String value : values) {
+            builder.append(value.getBytes(UTF_8));
+        }
+
+        return builder.build();
+    }
+
+    public static ValueBlock createBlock(long position, long... values)
+    {
+        BlockBuilder builder = new BlockBuilder(position, new TupleInfo(FIXED_INT_64));
+
+        for (long value : values) {
+            builder.append(value);
+        }
+
+        return builder.build();
+    }
+}

src/test/java/com/facebook/presto/TestAggregations.java

@@ -97,11 +97,11 @@ public AggregationFunction get()
     public Iterator<ValueBlock> newGroupColumn()
     {
         Iterator<ValueBlock> values = ImmutableList.<ValueBlock>builder()
-                .add(createBlock(0, "apple", "apple", "apple", "apple", "banana", "banana"))
-                .add(createBlock(20, "banana", "banana", "banana", "cherry", "cherry", "cherry"))
-                .add(createBlock(30, "date"))
-                .add(createBlock(31, "date"))
-                .add(createBlock(32, "date"))
+                .add(Blocks.createBlock(0, "apple", "apple", "apple", "apple", "banana", "banana"))
+                .add(Blocks.createBlock(20, "banana", "banana", "banana", "cherry", "cherry", "cherry"))
+                .add(Blocks.createBlock(30, "date"))
+                .add(Blocks.createBlock(31, "date"))
+                .add(Blocks.createBlock(32, "date"))
                 .build()
                 .iterator();
 
@@ -111,39 +111,17 @@ public Iterator<ValueBlock> newGroupColumn()
     public Iterator<ValueBlock> newAggregateColumn()
     {
         Iterator<ValueBlock> values = ImmutableList.<ValueBlock>builder()
-                .add(createBlock(0, 1L, 2L, 3L, 4L, 5L, 6L))
-                .add(createBlock(20, 1L, 2L, 3L, 4L, 5L, 6L))
-                .add(createBlock(30, 1L))
-                .add(createBlock(31, 2L))
-                .add(createBlock(32, 3L))
+                .add(Blocks.createBlock(0, 1L, 2L, 3L, 4L, 5L, 6L))
+                .add(Blocks.createBlock(20, 1L, 2L, 3L, 4L, 5L, 6L))
+                .add(Blocks.createBlock(30, 1L))
+                .add(Blocks.createBlock(31, 2L))
+                .add(Blocks.createBlock(32, 3L))
                 .build()
                 .iterator();
 
         return values;
     }
 
-    private ValueBlock createBlock(int position, String... values)
-    {
-        BlockBuilder builder = new BlockBuilder(position, new TupleInfo(VARIABLE_BINARY));
-
-        for (String value : values) {
-            builder.append(value.getBytes(UTF_8));
-        }
-
-        return builder.build();
-    }
-
-    private ValueBlock createBlock(long position, long... values)
-    {
-        BlockBuilder builder = new BlockBuilder(position, new TupleInfo(FIXED_INT_64));
-
-        for (long value : values) {
-            builder.append(value);
-        }
-
-        return builder.build();
-    }
-
     private Tuple createTuple(String key, long count)
     {
         TupleInfo tupleInfo = new TupleInfo(VARIABLE_BINARY, FIXED_INT_64);

src/test/java/com/facebook/presto/TestCountAggregation.java

@@ -0,0 +1,58 @@
+package com.facebook.presto;
+
+import com.facebook.presto.aggregations.CountAggregation;
+import org.testng.annotations.Test;
+
+import static com.facebook.presto.Blocks.createBlock;
+import static org.testng.Assert.assertEquals;
+
+public class TestCountAggregation
+{
+    @Test
+    public void testBasic()
+            throws Exception
+    {
+        ValueBlock values = createBlock(0, "apple", "banana", "cherry", "date");
+        PositionBlock positions = new RangePositionBlock(Range.create(0, 3));
+
+        assertCount(values, positions, 4);
+    }
+
+    @Test
+    public void testSubset()
+            throws Exception
+    {
+        ValueBlock values = createBlock(0, "apple", "banana", "cherry", "date");
+        PositionBlock positions = new RangePositionBlock(Range.create(1, 2));
+
+        assertCount(values, positions, 2);
+    }
+
+    @Test
+    public void testNonOverlapping()
+            throws Exception
+    {
+        ValueBlock values = createBlock(0, "apple", "banana", "cherry", "date");
+        PositionBlock positions = new RangePositionBlock(Range.create(10, 20));
+
+        assertCount(values, positions, 0);
+    }
+
+    @Test
+    public void testSparse()
+            throws Exception
+    {
+        ValueBlock values = createBlock(0, "apple", "banana", "cherry", "date");
+        PositionBlock positions = new UncompressedPositionBlock(1, 3);
+
+        assertCount(values, positions, 2);
+    }
+
+    private void assertCount(ValueBlock values, PositionBlock positions, int expected)
+    {
+        CountAggregation count = new CountAggregation();
+        count.add(values, positions);
+
+        assertEquals(count.evaluate().getLong(0), expected);
+    }
+}

src/test/java/com/facebook/presto/TestMaskedValueBlock.java

@@ -20,7 +20,7 @@ public class TestMaskedValueBlock
     public void testMaskUncompressedBlock()
             throws Exception
     {
-        ValueBlock valueBlock = createBlock(10, "alice", "bob", "charlie", "david", "eric", "frank", "greg", "hank", "ian", "jenny");
+        ValueBlock valueBlock = Blocks.createBlock(10, "alice", "bob", "charlie", "david", "eric", "frank", "greg", "hank", "ian", "jenny");
         Optional<ValueBlock> masked = MaskedValueBlock.maskBlock(valueBlock, new UncompressedPositionBlock(8, 10, 12, 14, 16, 100));
 
         assertTrue(masked.isPresent());
@@ -87,28 +87,4 @@ private Tuple createTuple(String value)
         TupleInfo tupleInfo = new TupleInfo(VARIABLE_BINARY);
         return tupleInfo.builder().append(Slices.copiedBuffer(value, UTF_8)).build();
     }
-
-    private ValueBlock createBlock(int position, String... values)
-    {
-        BlockBuilder builder = new BlockBuilder(position, new TupleInfo(VARIABLE_BINARY));
-
-        for (String value : values) {
-            builder.append(value.getBytes(UTF_8));
-        }
-
-        return builder.build();
-    }
-
-    private ValueBlock createBlock(long position, long... values)
-    {
-        BlockBuilder builder = new BlockBuilder(position, new TupleInfo(FIXED_INT_64));
-
-        for (long value : values) {
-            builder.append(value);
-        }
-
-        return builder.build();
-    }
-
-
 }