Refactor ColumnSelectorFactory; Rely on ColumnValueSelector's polymorphism

api/src/main/java/io/druid/data/input/MapBasedRow.java

@@ -22,22 +22,23 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.collect.Lists;
+import com.google.common.primitives.Longs;
 import io.druid.guice.annotations.PublicApi;
 import io.druid.java.util.common.DateTimes;
+import io.druid.java.util.common.StringUtils;
 import io.druid.java.util.common.parsers.ParseException;
 import org.joda.time.DateTime;
 
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.regex.Pattern;
 
 /**
  */
 @PublicApi
 public class MapBasedRow implements Row
 {
-  private static final Pattern LONG_PAT = Pattern.compile("[-|+]?\\d+");
+  private static final Long LONG_ZERO = 0L;
 
   private final DateTime timestamp;
   private final Map<String, Object> event;
@@ -101,19 +102,28 @@ public Object getRaw(String dimension)
   }
 
   @Override
-  public float getFloatMetric(String metric)
+  public Number getMetric(String metric)
   {
     Object metricValue = event.get(metric);
 
     if (metricValue == null) {
-      return 0.0f;
+      return LONG_ZERO;
     }
 
     if (metricValue instanceof Number) {
-      return ((Number) metricValue).floatValue();
+      return (Number) metricValue;
     } else if (metricValue instanceof String) {
       try {
-        return Float.valueOf(((String) metricValue).replace(",", ""));
+        String metricValueString = StringUtils.removeChar(((String) metricValue).trim(), ',');
+        // Longs.tryParse() doesn't support leading '+', so we need to trim it ourselves
+        metricValueString = trimLeadingPlusOfLongString(metricValueString);
+        Long v = Longs.tryParse(metricValueString);
+        // Do NOT use ternary operator here, because it makes Java to convert Long to Double
+        if (v != null) {
+          return v;
+        } else {
+          return Double.valueOf(metricValueString);
+        }
       }
       catch (Exception e) {
         throw new ParseException(e, "Unable to parse metrics[%s], value[%s]", metric, metricValue);
@@ -123,51 +133,15 @@ public float getFloatMetric(String metric)
     }
   }
 
-  @Override
-  public long getLongMetric(String metric)
+  private static String trimLeadingPlusOfLongString(String metricValueString)
   {
-    Object metricValue = event.get(metric);
-
-    if (metricValue == null) {
-      return 0L;
-    }
-
-    if (metricValue instanceof Number) {
-      return ((Number) metricValue).longValue();
-    } else if (metricValue instanceof String) {
-      try {
-        String s = ((String) metricValue).replace(",", "");
-        return LONG_PAT.matcher(s).matches() ? Long.valueOf(s) : Double.valueOf(s).longValue();
+    if (metricValueString.length() > 1 && metricValueString.charAt(0) == '+') {
+      char secondChar = metricValueString.charAt(1);
+      if (secondChar >= '0' && secondChar <= '9') {
+        metricValueString = metricValueString.substring(1);
       }
-      catch (Exception e) {
-        throw new ParseException(e, "Unable to parse metrics[%s], value[%s]", metric, metricValue);
-      }
-    } else {
-      throw new ParseException("Unknown type[%s]", metricValue.getClass());
-    }
-  }
-
-  @Override
-  public double getDoubleMetric(String metric)
-  {
-    Object metricValue = event.get(metric);
-
-    if (metricValue == null) {
-      return 0.0d;
-    }
-
-    if (metricValue instanceof Number) {
-      return ((Number) metricValue).doubleValue();
-    } else if (metricValue instanceof String) {
-      try {
-        return Double.valueOf(((String) metricValue).replace(",", ""));
-      }
-      catch (Exception e) {
-        throw new ParseException(e, "Unable to parse metrics[%s], value[%s]", metric, metricValue);
-      }
-    } else {
-      throw new ParseException("Unknown type[%s]", metricValue.getClass());
     }
+    return metricValueString;
   }
 
   @Override

api/src/main/java/io/druid/data/input/Row.java

@@ -64,7 +64,7 @@ public interface Row extends Comparable<Row>
   List<String> getDimension(String dimension);
 
   /**
-   * Returns the raw dimension value for the given column name. This is different from #getDimension which
+   * Returns the raw dimension value for the given column name. This is different from {@link #getDimension} which
    * all values to strings before returning them.
    *
    * @param dimension the column name of the dimension requested
@@ -74,32 +74,10 @@ public interface Row extends Comparable<Row>
   Object getRaw(String dimension);
 
   /**
-   * Returns the float value of the given metric column.
-   * <p/>
-   *
-   * @param metric the column name of the metric requested
-   *
-   * @return the float value for the provided column name.
-   */
-  float getFloatMetric(String metric);
-
-  /**
-   * Returns the long value of the given metric column.
-   * <p/>
-   *
-   * @param metric the column name of the metric requested
-   *
-   * @return the long value for the provided column name.
-   */
-  long getLongMetric(String metric);
-
-  /**
-   * Returns the double value of the given metric column.
-   * <p/>
-   *
-   * @param metric the column name of the metric requested
-   *
-   * @return the double value for the provided column name.
+   * Returns the metric column value for the given column name. This method is different from {@link #getRaw} in two
+   * aspects:
+   *  1. If the column is absent in the row, numeric zero is returned, rather than null.
+   *  2. If the column has string value, an attempt is made to parse this value as a number.
    */
-  double getDoubleMetric(String metric);
+  Number getMetric(String metric);
 }

api/src/test/java/io/druid/data/input/MapBasedRowTest.java

@@ -42,12 +42,12 @@ public void testGetLongMetricFromString()
           .build()
     );
 
-    Assert.assertEquals(-1, row.getLongMetric("k0"));
-    Assert.assertEquals(1, row.getLongMetric("k1"));
-    Assert.assertEquals(1, row.getLongMetric("k2"));
-    Assert.assertEquals(100000, row.getLongMetric("k3"));
-    Assert.assertEquals(9223372036854775806L, row.getLongMetric("k4"));
-    Assert.assertEquals(-9223372036854775807L, row.getLongMetric("k5"));
-    Assert.assertEquals(9223372036854775802L, row.getLongMetric("k6"));
+    Assert.assertEquals(-1.2, row.getMetric("k0"));
+    Assert.assertEquals(1.23, row.getMetric("k1"));
+    Assert.assertEquals(1.8, row.getMetric("k2"));
+    Assert.assertEquals(100000.0, row.getMetric("k3"));
+    Assert.assertEquals(9223372036854775806L, row.getMetric("k4"));
+    Assert.assertEquals(-9223372036854775807L, row.getMetric("k5"));
+    Assert.assertEquals(9223372036854775802L, row.getMetric("k6"));
   }
 }

api/src/test/java/io/druid/data/input/impl/InputRowParserSerdeTest.java

@@ -145,11 +145,11 @@ public void testMapInputRowParserNumbersSerde() throws Exception
         ImmutableList.of("1412705931123", "123.456", "1.23E47", "hello"),
         parsed.getDimension("values")
     );
-    Assert.assertEquals(Float.POSITIVE_INFINITY, parsed.getFloatMetric("toobig"), 0.0);
+    Assert.assertEquals(Float.POSITIVE_INFINITY, parsed.getMetric("toobig").floatValue(), 0.0);
     Assert.assertEquals(123E64, parsed.getRaw("toobig"));
-    Assert.assertEquals(123.456f, parsed.getFloatMetric("value"), 0.0f);
+    Assert.assertEquals(123.456f, parsed.getMetric("value").floatValue(), 0.0f);
     Assert.assertEquals(123456789000L, parsed.getRaw("long"));
-    Assert.assertEquals(1.23456791E11f, parsed.getFloatMetric("long"), 0.0f);
+    Assert.assertEquals(1.23456791E11f, parsed.getMetric("long").floatValue(), 0.0f);
     Assert.assertEquals(1412705931123L, parsed.getTimestampFromEpoch());
   }
 

benchmarks/src/main/java/io/druid/benchmark/ExpressionBenchmark.java

@@ -33,9 +33,9 @@
 import io.druid.query.aggregation.DoubleSumAggregatorFactory;
 import io.druid.query.aggregation.JavaScriptAggregatorFactory;
 import io.druid.query.expression.TestExprMacroTable;
+import io.druid.segment.BaseFloatColumnValueSelector;
 import io.druid.segment.ColumnSelectorFactory;
 import io.druid.segment.Cursor;
-import io.druid.segment.FloatColumnSelector;
 import io.druid.segment.QueryableIndex;
 import io.druid.segment.QueryableIndexStorageAdapter;
 import io.druid.segment.VirtualColumns;
@@ -152,8 +152,8 @@ public void queryUsingNative(Blackhole blackhole) throws Exception
     final Double result = compute(
         columnSelectorFactory ->
             new NativeBufferAggregator(
-                columnSelectorFactory.makeFloatColumnSelector("x"),
-                columnSelectorFactory.makeFloatColumnSelector("y")
+                columnSelectorFactory.makeColumnValueSelector("x"),
+                columnSelectorFactory.makeColumnValueSelector("y")
             )
     );
     blackhole.consume(result);
@@ -197,10 +197,13 @@ private double compute(final Function<ColumnSelectorFactory, BufferAggregator> a
 
   private static class NativeBufferAggregator implements BufferAggregator
   {
-    private final FloatColumnSelector xSelector;
-    private final FloatColumnSelector ySelector;
+    private final BaseFloatColumnValueSelector xSelector;
+    private final BaseFloatColumnValueSelector ySelector;
 
-    public NativeBufferAggregator(final FloatColumnSelector xSelector, final FloatColumnSelector ySelector)
+    public NativeBufferAggregator(
+        final BaseFloatColumnValueSelector xSelector,
+        final BaseFloatColumnValueSelector ySelector
+    )
     {
       this.xSelector = xSelector;
       this.ySelector = ySelector;

benchmarks/src/main/java/io/druid/benchmark/FilterPartitionBenchmark.java

@@ -52,12 +52,12 @@
 import io.druid.query.filter.OrDimFilter;
 import io.druid.query.filter.SelectorDimFilter;
 import io.druid.query.ordering.StringComparators;
+import io.druid.segment.BaseLongColumnValueSelector;
 import io.druid.segment.Cursor;
 import io.druid.segment.DimensionSelector;
 import io.druid.segment.IndexIO;
 import io.druid.segment.IndexMergerV9;
 import io.druid.segment.IndexSpec;
-import io.druid.segment.LongColumnSelector;
 import io.druid.segment.QueryableIndex;
 import io.druid.segment.QueryableIndexStorageAdapter;
 import io.druid.segment.StorageAdapter;
@@ -542,7 +542,7 @@ private Sequence<List<Long>> readCursorsLong(Sequence<Cursor> cursors, final Bla
           public List<Long> apply(Cursor input)
           {
             List<Long> longvals = new ArrayList<Long>();
-            LongColumnSelector selector = input.getColumnSelectorFactory().makeLongColumnSelector("sumLongSequential");
+            BaseLongColumnValueSelector selector = input.getColumnSelectorFactory().makeColumnValueSelector("sumLongSequential");
             while (!input.isDone()) {
               long rowval = selector.getLong();
               blackhole.consume(rowval);

extensions-core/datasketches/src/main/java/io/druid/query/aggregation/datasketches/theta/EmptySketchAggregator.java → common/src/main/java/io/druid/annotations/EverythingIsNonnullByDefault.java

@@ -17,52 +17,24 @@
  * under the License.
  */
 
-package io.druid.query.aggregation.datasketches.theta;
-
-import io.druid.query.aggregation.Aggregator;
-
-public class EmptySketchAggregator implements Aggregator
+package io.druid.annotations;
+
+import javax.annotation.Nonnull;
+import javax.annotation.meta.TypeQualifierDefault;
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+
+/**
+ * This annotation can be applied to a package, class or method to indicate that all
+ * class fields and method parameters and return values in that element are nonnull
+ * by default unless overridden.
+ */
+@Documented
+@Nonnull
+@TypeQualifierDefault({ElementType.FIELD, ElementType.METHOD, ElementType.PARAMETER})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface EverythingIsNonnullByDefault
 {
-  public EmptySketchAggregator()
-  {
-  }
-
-  @Override
-  public void aggregate()
-  {
-  }
-
-  @Override
-  public void reset()
-  {
-  }
-
-  @Override
-  public Object get()
-  {
-    return SketchHolder.EMPTY;
-  }
-
-  @Override
-  public float getFloat()
-  {
-    throw new UnsupportedOperationException("Not implemented");
-  }
-
-  @Override
-  public long getLong()
-  {
-    throw new UnsupportedOperationException("Not implemented");
-  }
-
-  @Override
-  public double getDouble()
-  {
-    throw new UnsupportedOperationException("Not implemented");
-  }
-
-  @Override
-  public void close()
-  {
-  }
 }

common/src/main/java/io/druid/common/utils/UUIDUtils.java

@@ -21,6 +21,7 @@
 
 import com.google.common.base.Joiner;
 import com.google.common.base.Strings;
+import io.druid.java.util.common.StringUtils;
 
 import java.util.ArrayList;
 import java.util.UUID;
@@ -53,7 +54,8 @@ public static String generateUuid(String... extraData)
         extra = Joiner.on(UUID_DELIM).join(extraStrings);
       }
     }
-    final String uuid = UUID.randomUUID().toString().replace("-", ""); // We don't use "-" in general, so remove them here.
+    // We don't use "-" in general, so remove them here.
+    final String uuid = StringUtils.removeChar(UUID.randomUUID().toString(), '-');
     return extra == null ? uuid : (extra + UUID_DELIM + uuid);
   }
 }

extensions-contrib/time-min-max/src/main/java/io/druid/query/aggregation/TimestampAggregator.java

@@ -20,7 +20,7 @@
 package io.druid.query.aggregation;
 
 import io.druid.data.input.impl.TimestampSpec;
-import io.druid.segment.ObjectColumnSelector;
+import io.druid.segment.BaseObjectColumnValueSelector;
 
 import java.util.Comparator;
 
@@ -37,7 +37,7 @@ static Object combineValues(Comparator<Long> comparator, Object lhs, Object rhs)
     }
   }
 
-  private final ObjectColumnSelector selector;
+  private final BaseObjectColumnValueSelector selector;
   private final String name;
   private final TimestampSpec timestampSpec;
   private final Comparator<Long> comparator;
@@ -47,7 +47,7 @@ static Object combineValues(Comparator<Long> comparator, Object lhs, Object rhs)
 
   public TimestampAggregator(
       String name,
-      ObjectColumnSelector selector,
+      BaseObjectColumnValueSelector selector,
       TimestampSpec timestampSpec,
       Comparator<Long> comparator,
       Long initValue