Introduce "transformSpec" at ingest-time.

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

@@ -21,15 +21,10 @@
 
 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;
 
@@ -38,8 +33,6 @@
 @PublicApi
 public class MapBasedRow implements Row
 {
-  private static final Long LONG_ZERO = 0L;
-
   private final DateTime timestamp;
   private final Map<String, Object> event;
 
@@ -83,16 +76,7 @@ public Map<String, Object> getEvent()
   @Override
   public List<String> getDimension(String dimension)
   {
-    final Object dimValue = event.get(dimension);
-
-    if (dimValue == null) {
-      return Collections.emptyList();
-    } else if (dimValue instanceof List) {
-      // guava's toString function fails on null objects, so please do not use it
-      return Lists.transform((List) dimValue, String::valueOf);
-    } else {
-      return Collections.singletonList(String.valueOf(dimValue));
-    }
+    return Rows.objectToStrings(event.get(dimension));
   }
 
   @Override
@@ -104,44 +88,7 @@ public Object getRaw(String dimension)
   @Override
   public Number getMetric(String metric)
   {
-    Object metricValue = event.get(metric);
-
-    if (metricValue == null) {
-      return LONG_ZERO;
-    }
-
-    if (metricValue instanceof Number) {
-      return (Number) metricValue;
-    } else if (metricValue instanceof String) {
-      try {
-        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);
-      }
-    } else {
-      throw new ParseException("Unknown type[%s]", metricValue.getClass());
-    }
-  }
-
-  private static String trimLeadingPlusOfLongString(String metricValueString)
-  {
-    if (metricValueString.length() > 1 && metricValueString.charAt(0) == '+') {
-      char secondChar = metricValueString.charAt(1);
-      if (secondChar >= '0' && secondChar <= '9') {
-        metricValueString = metricValueString.substring(1);
-      }
-    }
-    return metricValueString;
+    return Rows.objectToNumber(metric, event.get(metric));
   }
 
   @Override

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

@@ -22,7 +22,12 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSortedSet;
 import com.google.common.collect.Maps;
+import com.google.common.primitives.Longs;
+import io.druid.java.util.common.StringUtils;
+import io.druid.java.util.common.parsers.ParseException;
 
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -31,9 +36,12 @@
  */
 public class Rows
 {
+  public static final Long LONG_ZERO = 0L;
+
   /**
    * @param timeStamp rollup up timestamp to be used to create group key
-   * @param inputRow input row
+   * @param inputRow  input row
+   *
    * @return groupKey for the given input row
    */
   public static List<Object> toGroupKey(long timeStamp, InputRow inputRow)
@@ -50,4 +58,77 @@ public static List<Object> toGroupKey(long timeStamp, InputRow inputRow)
         dims
     );
   }
+
+  /**
+   * Convert an object to a list of strings.
+   */
+  public static List<String> objectToStrings(final Object inputValue)
+  {
+    if (inputValue == null) {
+      return Collections.emptyList();
+    } else if (inputValue instanceof List) {
+      // guava's toString function fails on null objects, so please do not use it
+      final List<Object> values = (List) inputValue;
+
+      final List<String> retVal = new ArrayList<>(values.size());
+      for (Object val : values) {
+        retVal.add(String.valueOf(val));
+      }
+
+      return retVal;
+    } else {
+      return Collections.singletonList(String.valueOf(inputValue));
+    }
+  }
+
+  /**
+   * Convert an object to a number. Nulls are treated as zeroes.
+   *
+   * @param name       field name of the object being converted (may be used for exception messages)
+   * @param inputValue the actual object being converted
+   *
+   * @return a number
+   *
+   * @throws NullPointerException if the string is null
+   * @throws ParseException       if the column cannot be converted to a number
+   */
+  public static Number objectToNumber(final String name, final Object inputValue)
+  {
+    if (inputValue == null) {
+      return Rows.LONG_ZERO;
+    }
+
+    if (inputValue instanceof Number) {
+      return (Number) inputValue;
+    } else if (inputValue instanceof String) {
+      try {
+        String metricValueString = StringUtils.removeChar(((String) inputValue).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 value[%s] for field[%s]", inputValue, name);
+      }
+    } else {
+      throw new ParseException("Unknown type[%s] for field", inputValue.getClass(), inputValue);
+    }
+  }
+
+  private static String trimLeadingPlusOfLongString(String metricValueString)
+  {
+    if (metricValueString.length() > 1 && metricValueString.charAt(0) == '+') {
+      char secondChar = metricValueString.charAt(1);
+      if (secondChar >= '0' && secondChar <= '9') {
+        metricValueString = metricValueString.substring(1);
+      }
+    }
+    return metricValueString;
+  }
 }

api/src/main/java/io/druid/data/input/impl/DimensionsSpec.java

@@ -121,7 +121,8 @@ public Set<String> getDimensionExclusions()
     return dimensionExclusions;
   }
 
-  @Deprecated @JsonIgnore
+  @Deprecated
+  @JsonIgnore
   public List<SpatialDimensionSchema> getSpatialDimensions()
   {
     Iterable<NewSpatialDimensionSchema> filteredList = Iterables.filter(
@@ -244,4 +245,13 @@ public int hashCode()
     result = 31 * result + dimensionExclusions.hashCode();
     return result;
   }
+
+  @Override
+  public String toString()
+  {
+    return "DimensionsSpec{" +
+           "dimensions=" + dimensions +
+           ", dimensionExclusions=" + dimensionExclusions +
+           '}';
+  }
 }

api/src/main/java/io/druid/data/input/impl/FileIteratingFirehose.java

@@ -24,6 +24,7 @@
 import io.druid.utils.Runnables;
 import org.apache.commons.io.LineIterator;
 
+import javax.annotation.Nullable;
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Iterator;
@@ -69,6 +70,7 @@ public boolean hasMore()
     return lineIterator != null && lineIterator.hasNext();
   }
 
+  @Nullable
   @Override
   public InputRow nextRow()
   {

api/src/main/java/io/druid/data/input/impl/InputRowParser.java

@@ -24,6 +24,8 @@
 import io.druid.data.input.InputRow;
 import io.druid.guice.annotations.ExtensionPoint;
 
+import javax.annotation.Nullable;
+
 @ExtensionPoint
 @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type", defaultImpl = StringInputRowParser.class)
 @JsonSubTypes(value = {
@@ -33,6 +35,11 @@
 })
 public interface InputRowParser<T>
 {
+  /**
+   * Parse an input into an {@link InputRow}. Return null if this input should be thrown away, or throws
+   * {@code ParseException} if the input is unparseable.
+   */
+  @Nullable
   InputRow parse(T input);
 
   ParseSpec getParseSpec();

api/src/main/java/io/druid/data/input/impl/JSONParseSpec.java

@@ -30,6 +30,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 /**
  */
@@ -97,4 +98,38 @@ public Map<String, Boolean> getFeatureSpec()
   {
     return featureSpec;
   }
+
+  @Override
+  public boolean equals(final Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    if (!super.equals(o)) {
+      return false;
+    }
+    final JSONParseSpec that = (JSONParseSpec) o;
+    return Objects.equals(flattenSpec, that.flattenSpec) &&
+           Objects.equals(featureSpec, that.featureSpec);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(super.hashCode(), flattenSpec, featureSpec);
+  }
+
+  @Override
+  public String toString()
+  {
+    return "JSONParseSpec{" +
+           "timestampSpec=" + getTimestampSpec() +
+           ", dimensionsSpec=" + getDimensionsSpec() +
+           ", flattenSpec=" + flattenSpec +
+           ", featureSpec=" + featureSpec +
+           '}';
+  }
 }

docs/content/misc/math-expr.md

@@ -13,13 +13,16 @@ This expression language supports the following operators (listed in decreasing
 |<, <=, >, >=, ==, !=|Binary Comparison|
 |&&,\|\||Binary Logical AND, OR|
 
-Long, double and string data types are supported. If a number contains a dot, it is interpreted as a double, otherwise it is interpreted as a long. That means, always add a '.' to your number if you want it interpreted as a double value. String literal should be quoted by single quotation marks.
+Long, double, and string data types are supported. If a number contains a dot, it is interpreted as a double, otherwise it is interpreted as a long. That means, always add a '.' to your number if you want it interpreted as a double value. String literals should be quoted by single quotation marks.
 
-Expressions can contain variables. Variable names may contain letters, digits, '\_' and '$'. Variable names must not begin with a digit. To escape other special characters, user can quote it with double quotation marks.
+Multi-value types are not fully supported yet. Expressions may behave inconsistently on multi-value types, and you
+should not rely on the behavior in this case to stay the same in future releases.
 
-For logical operators, a number is true if and only if it is positive (0 or minus value means false). For string type, it's evaluation result of 'Boolean.valueOf(string)'.
+Expressions can contain variables. Variable names may contain letters, digits, '\_' and '$'. Variable names must not begin with a digit. To escape other special characters, you can quote it with double quotation marks.
 
-Also, the following built-in functions are supported.
+For logical operators, a number is true if and only if it is positive (0 or negative value means false). For string type, it's the evaluation result of 'Boolean.valueOf(string)'.
+
+The following built-in functions are available.
 
 ## General functions
 

examples/src/main/java/io/druid/examples/twitter/TwitterSpritzerFirehoseFactory.java

@@ -250,6 +250,7 @@ public boolean hasMore()
         }
       }
 
+      @Nullable
       @Override
       public InputRow nextRow()
       {

extensions-contrib/druid-rocketmq/src/main/java/io/druid/firehose/rocketmq/RocketMQFirehoseFactory.java

@@ -34,14 +34,15 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.collect.Sets;
-import io.druid.data.input.ByteBufferInputRowParser;
 import io.druid.data.input.Firehose;
 import io.druid.data.input.FirehoseFactory;
 import io.druid.data.input.InputRow;
+import io.druid.data.input.impl.InputRowParser;
 import io.druid.java.util.common.StringUtils;
 import io.druid.java.util.common.logger.Logger;
 import io.druid.java.util.common.parsers.ParseException;
 
+import javax.annotation.Nullable;
 import java.io.File;
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -57,7 +58,7 @@
 import java.util.concurrent.ConcurrentSkipListSet;
 import java.util.concurrent.CountDownLatch;
 
-public class RocketMQFirehoseFactory implements FirehoseFactory<ByteBufferInputRowParser>
+public class RocketMQFirehoseFactory implements FirehoseFactory<InputRowParser<ByteBuffer>>
 {
 
   private static final Logger LOGGER = new Logger(RocketMQFirehoseFactory.class);
@@ -139,7 +140,7 @@ private boolean hasMessagesPending()
 
   @Override
   public Firehose connect(
-      ByteBufferInputRowParser byteBufferInputRowParser,
+      InputRowParser<ByteBuffer> byteBufferInputRowParser,
       File temporaryDirectory
   ) throws IOException, ParseException
   {
@@ -149,7 +150,7 @@ public Firehose connect(
         Sets.newHashSet("feed")
     );
 
-    final ByteBufferInputRowParser theParser = byteBufferInputRowParser.withParseSpec(
+    final InputRowParser<ByteBuffer> theParser = byteBufferInputRowParser.withParseSpec(
         byteBufferInputRowParser.getParseSpec()
                                 .withDimensionsSpec(
                                     byteBufferInputRowParser.getParseSpec()
@@ -247,6 +248,7 @@ public boolean hasMore()
         return hasMore;
       }
 
+      @Nullable
       @Override
       public InputRow nextRow()
       {

extensions-contrib/orc-extensions/src/test/java/io/druid/data/input/orc/DruidOrcInputFormatTest.java

@@ -19,6 +19,7 @@
 package io.druid.data.input.orc;
 
 import io.druid.data.input.MapBasedInputRow;
+import io.druid.data.input.impl.InputRowParser;
 import io.druid.indexer.HadoopDruidIndexerConfig;
 import io.druid.java.util.common.DateTimes;
 import io.druid.java.util.common.StringUtils;
@@ -90,7 +91,7 @@ public void testRead() throws IOException, InterruptedException
 
     TaskAttemptContext context = new TaskAttemptContextImpl(job.getConfiguration(), new TaskAttemptID());
     RecordReader reader = inputFormat.createRecordReader(split, context);
-    OrcHadoopInputRowParser parser = (OrcHadoopInputRowParser) config.getParser();
+    InputRowParser<OrcStruct> parser = (InputRowParser<OrcStruct>) config.getParser();
 
     reader.initialize(split, context);