[WIP] support for double and int columns

extensions/datasketches/src/main/java/io/druid/query/aggregation/datasketches/theta/EmptySketchAggregator.java

@@ -58,6 +58,18 @@ public long getLong()
     throw new UnsupportedOperationException("Not implemented");
   }
 
+  @Override
+  public int getInt()
+  {
+    throw new UnsupportedOperationException("Not implemented");
+  }
+
+  @Override
+  public double getDouble()
+  {
+    throw new UnsupportedOperationException("Not implemented");
+  }
+
   @Override
   public String getName()
   {

extensions/datasketches/src/main/java/io/druid/query/aggregation/datasketches/theta/EmptySketchBufferAggregator.java

@@ -57,6 +57,18 @@ public long getLong(ByteBuffer buf, int position)
     throw new UnsupportedOperationException("Not implemented");
   }
 
+  @Override
+  public int getInt(ByteBuffer buf, int position)
+  {
+    throw new UnsupportedOperationException("Not implemented");
+  }
+
+  @Override
+  public double getDouble(ByteBuffer buf, int position)
+  {
+    throw new UnsupportedOperationException("Not implemented");
+  }
+
   @Override
   public void close()
   {

extensions/datasketches/src/main/java/io/druid/query/aggregation/datasketches/theta/SketchAggregator.java

@@ -89,6 +89,18 @@ public long getLong()
     throw new UnsupportedOperationException("Not implemented");
   }
 
+  @Override
+  public int getInt()
+  {
+    throw new UnsupportedOperationException("Not implemented");
+  }
+
+  @Override
+  public double getDouble()
+  {
+    throw new UnsupportedOperationException("Not implemented");
+  }
+
   @Override
   public String getName()
   {

extensions/datasketches/src/main/java/io/druid/query/aggregation/datasketches/theta/SketchAggregatorFactory.java

@@ -159,6 +159,12 @@ public Object getAggregatorStartValue()
     return Sketches.updateSketchBuilder().build(size);
   }
 
+  @Override
+  public boolean isComplex()
+  {
+    return true;
+  }
+
   @Override
   public List<String> requiredFields()
   {

extensions/datasketches/src/main/java/io/druid/query/aggregation/datasketches/theta/SketchBufferAggregator.java

@@ -110,6 +110,18 @@ public long getLong(ByteBuffer buf, int position)
     throw new UnsupportedOperationException("Not implemented");
   }
 
+  @Override
+  public int getInt(ByteBuffer buf, int position)
+  {
+    throw new UnsupportedOperationException("Not implemented");
+  }
+
+  @Override
+  public double getDouble(ByteBuffer buf, int position)
+  {
+    throw new UnsupportedOperationException("Not implemented");
+  }
+
   @Override
   public void close()
   {

extensions/histogram/src/main/java/io/druid/query/aggregation/histogram/ApproximateHistogramAggregator.java

@@ -95,6 +95,18 @@ public long getLong()
     throw new UnsupportedOperationException("ApproximateHistogramAggregator does not support getLong()");
   }
 
+  @Override
+  public double getDouble()
+  {
+    throw new UnsupportedOperationException("ApproximateHistogramAggregator does not support getDouble()");
+  }
+
+  @Override
+  public int getInt()
+  {
+    throw new UnsupportedOperationException("ApproximateHistogramAggregator does not support getInt()");
+  }
+
   @Override
   public String getName()
   {

extensions/histogram/src/main/java/io/druid/query/aggregation/histogram/ApproximateHistogramAggregatorFactory.java

@@ -238,6 +238,12 @@ public Object getAggregatorStartValue()
     return new ApproximateHistogram(resolution);
   }
 
+  @Override
+  public boolean isComplex()
+  {
+    return true;
+  }
+
   @Override
   public boolean equals(Object o)
   {

extensions/histogram/src/main/java/io/druid/query/aggregation/histogram/ApproximateHistogramBufferAggregator.java

@@ -94,6 +94,18 @@ public long getLong(ByteBuffer buf, int position)
     throw new UnsupportedOperationException("ApproximateHistogramBufferAggregator does not support getLong()");
   }
 
+  @Override
+  public int getInt(ByteBuffer buf, int position)
+  {
+    throw new UnsupportedOperationException("ApproximateHistogramBufferAggregator does not support getInt()");
+  }
+
+  @Override
+  public double getDouble(ByteBuffer buf, int position)
+  {
+    throw new UnsupportedOperationException("ApproximateHistogramBufferAggregator does not support getDouble()");
+  }
+
   @Override
   public void close()
   {

extensions/histogram/src/main/java/io/druid/query/aggregation/histogram/ApproximateHistogramFoldingAggregator.java

@@ -93,6 +93,18 @@ public long getLong()
     throw new UnsupportedOperationException("ApproximateHistogramFoldingAggregator does not support getLong()");
   }
 
+  @Override
+  public int getInt()
+  {
+    throw new UnsupportedOperationException("ApproximateHistogramFoldingAggregator does not support getInt()");
+  }
+
+  @Override
+  public double getDouble()
+  {
+    throw new UnsupportedOperationException("ApproximateHistogramFoldingAggregator does not support getDouble()");
+  }
+
   @Override
   public String getName()
   {

extensions/histogram/src/main/java/io/druid/query/aggregation/histogram/ApproximateHistogramFoldingBufferAggregator.java

@@ -97,6 +97,18 @@ public long getLong(ByteBuffer buf, int position)
     throw new UnsupportedOperationException("ApproximateHistogramFoldingBufferAggregator does not support getLong()");
   }
 
+  @Override
+  public int getInt(ByteBuffer buf, int position)
+  {
+    throw new UnsupportedOperationException("ApproximateHistogramFoldingBufferAggregator does not support getInt()");
+  }
+
+  @Override
+  public double getDouble(ByteBuffer buf, int position)
+  {
+    throw new UnsupportedOperationException("ApproximateHistogramFoldingBufferAggregator does not support getDouble()");
+  }
+
   @Override
   public void close()
   {

indexing-hadoop/src/main/java/io/druid/indexer/IndexGeneratorJob.java

@@ -397,6 +397,18 @@ public long getLongMetric(String metric)
           return row.getLongMetric(metric);
         }
 
+        @Override
+        public int getIntMetric(String metric)
+        {
+          return row.getIntMetric(metric);
+        }
+
+        @Override
+        public double getDoubleMetric(String metric)
+        {
+          return row.getDoubleMetric(metric);
+        }
+
         @Override
         public int compareTo(Row o)
         {

indexing-hadoop/src/main/java/io/druid/indexer/InputRowSerde.java

@@ -99,6 +99,10 @@ public InputRow get()
           out.writeFloat(agg.getFloat());
         } else if (t.equals("long")) {
           WritableUtils.writeVLong(out, agg.getLong());
+        } else if (t.equals("int")) {
+          out.writeInt(agg.getInt());
+        } else if (t.equals("double")) {
+          out.writeDouble(agg.getDouble());
         } else {
           //its a complex metric
           Object val = agg.get();
@@ -199,6 +203,10 @@ public static final InputRow fromBytes(byte[] data, AggregatorFactory[] aggs)
           event.put(metric, in.readFloat());
         } else if (type.equals("long")) {
           event.put(metric, WritableUtils.readVLong(in));
+        } else if (type.equals("int")) {
+          event.put(metric, in.readInt());
+        } else if (type.equals("double")) {
+          event.put(metric, in.readDouble());
         } else {
           ComplexMetricSerde serde = getComplexMetricSerde(type);
           byte[] value = readBytes(in);

indexing-hadoop/src/main/java/io/druid/indexer/hadoop/SegmentInputRow.java

@@ -81,6 +81,18 @@ public long getLongMetric(String metric)
     return delegate.getLongMetric(metric);
   }
 
+  @Override
+  public double getDoubleMetric(String metric)
+  {
+    return delegate.getDoubleMetric(metric);
+  }
+
+  @Override
+  public int getIntMetric(String metric)
+  {
+    return delegate.getIntMetric(metric);
+  }
+
   @Override
   public int compareTo(Row row)
   {

pom.xml

@@ -69,7 +69,7 @@
         <apache.curator.version>2.9.1</apache.curator.version>
         <jetty.version>9.2.5.v20141112</jetty.version>
         <jersey.version>1.19</jersey.version>
-        <druid.api.version>0.3.15</druid.api.version>
+        <druid.api.version>0.3.16-SNAPSHOT</druid.api.version>
         <!-- Watch out for Hadoop compatibility when updating to >= 2.5; see https://github.com/druid-io/druid/pull/1669 -->
         <jackson.version>2.4.6</jackson.version>
         <log4j.version>2.5</log4j.version>

processing/src/main/java/io/druid/jackson/AggregatorsModule.java

@@ -29,7 +29,9 @@
 import io.druid.query.aggregation.DoubleMinAggregatorFactory;
 import io.druid.query.aggregation.DoubleSumAggregatorFactory;
 import io.druid.query.aggregation.FilteredAggregatorFactory;
+import io.druid.query.aggregation.FloatSumAggregator;
 import io.druid.query.aggregation.HistogramAggregatorFactory;
+import io.druid.query.aggregation.IntSumAggregatorFactory;
 import io.druid.query.aggregation.JavaScriptAggregatorFactory;
 import io.druid.query.aggregation.LongMaxAggregatorFactory;
 import io.druid.query.aggregation.LongMinAggregatorFactory;
@@ -74,7 +76,9 @@ public AggregatorsModule()
       @JsonSubTypes.Type(name = "histogram", value = HistogramAggregatorFactory.class),
       @JsonSubTypes.Type(name = "hyperUnique", value = HyperUniquesAggregatorFactory.class),
       @JsonSubTypes.Type(name = "cardinality", value = CardinalityAggregatorFactory.class),
-      @JsonSubTypes.Type(name = "filtered", value = FilteredAggregatorFactory.class)
+      @JsonSubTypes.Type(name = "filtered", value = FilteredAggregatorFactory.class),
+      @JsonSubTypes.Type(name = "intSum", value = IntSumAggregatorFactory.class),
+      @JsonSubTypes.Type(name = "floatSum", value = FloatSumAggregator.class)
   })
   public static interface AggregatorFactoryMixin
   {

processing/src/main/java/io/druid/query/DruidMetrics.java

@@ -54,7 +54,7 @@ public static int findNumComplexAggs(List<AggregatorFactory> aggs)
     int retVal = 0;
     for (AggregatorFactory agg : aggs) {
       // This needs to change when we have support column types better
-      if (!agg.getTypeName().equals("float") && !agg.getTypeName().equals("long")) {
+      if (agg.isComplex()) {
         retVal++;
       }
     }

processing/src/main/java/io/druid/query/aggregation/Aggregator.java

@@ -37,8 +37,9 @@ public interface Aggregator {
   void reset();
   Object get();
   float getFloat();
+  long getLong();
+  int getInt();
+  double getDouble();
   String getName();
   void close();
-
-  long getLong();
 }

processing/src/main/java/io/druid/query/aggregation/AggregatorFactory.java

@@ -112,4 +112,10 @@ public interface AggregatorFactory
    * @return the starting value for a corresponding aggregator.
    */
   public Object getAggregatorStartValue();
+
+  /**
+   * Returns if the aggregator is doing complex or simple (primitive) aggregations
+   * @return if the aggregator is complex
+   */
+  public boolean isComplex();
 }

processing/src/main/java/io/druid/query/aggregation/Aggregators.java

@@ -53,6 +53,18 @@ public float getFloat()
         return 0;
       }
 
+      @Override
+      public double getDouble()
+      {
+        return 0;
+      }
+
+      @Override
+      public int getInt()
+      {
+        return 0;
+      }
+
       @Override
       public String getName()
       {
@@ -101,13 +113,24 @@ public float getFloat(ByteBuffer buf, int position)
         return 0;
       }
 
-
       @Override
       public long getLong(ByteBuffer buf, int position)
       {
         return 0L;
       }
 
+      @Override
+      public int getInt(ByteBuffer buf, int position)
+      {
+        return 0;
+      }
+
+      @Override
+      public double getDouble(ByteBuffer buf, int position)
+      {
+        return 0d;
+      }
+
       @Override
       public void close()
       {