Fix NPE when indexing unparseable/missing numeric values when sortFacts=false

processing/src/main/java/io/druid/segment/DimensionHandlerUtils.java

@@ -40,6 +40,11 @@
 
 public final class DimensionHandlerUtils
 {
+  // use these values to ensure that convertObjectToLong() and convertObjectToFloat() return the same
+  // boxed object when returning a constant zero
+  private final static Long LONG_ZERO = 0L;
+  private final static Float FLOAT_ZERO = 0.0f;
+
   private DimensionHandlerUtils() {}
 
   public final static ColumnCapabilities DEFAULT_STRING_CAPABILITIES =
@@ -225,15 +230,16 @@ private static <ColumnSelectorStrategyClass extends ColumnSelectorStrategy> Colu
   public static Long convertObjectToLong(Object valObj)
   {
     if (valObj == null) {
-      return 0L;
+      return LONG_ZERO;
     }
 
     if (valObj instanceof Long) {
       return (Long) valObj;
     } else if (valObj instanceof Number) {
       return ((Number) valObj).longValue();
     } else if (valObj instanceof String) {
-      return DimensionHandlerUtils.getExactLongFromDecimalString((String) valObj);
+      Long parsedVal = DimensionHandlerUtils.getExactLongFromDecimalString((String) valObj);
+      return parsedVal == null ? LONG_ZERO : parsedVal;
     } else {
       throw new ParseException("Unknown type[%s]", valObj.getClass());
     }
@@ -242,15 +248,16 @@ public static Long convertObjectToLong(Object valObj)
   public static Float convertObjectToFloat(Object valObj)
   {
     if (valObj == null) {
-      return 0.0f;
+      return FLOAT_ZERO;
     }
 
     if (valObj instanceof Float) {
       return (Float) valObj;
     } else if (valObj instanceof Number) {
       return ((Number) valObj).floatValue();
     } else if (valObj instanceof String) {
-      return Floats.tryParse((String) valObj);
+      Float parsedVal = Floats.tryParse((String) valObj);
+      return parsedVal == null ? FLOAT_ZERO : parsedVal;
     } else {
       throw new ParseException("Unknown type[%s]", valObj.getClass());
     }

processing/src/main/java/io/druid/segment/FloatDimensionIndexer.java

@@ -199,13 +199,16 @@ public int compareUnsortedEncodedKeyComponents(Float lhs, Float rhs)
   @Override
   public boolean checkUnsortedEncodedKeyComponentsEqual(Float lhs, Float rhs)
   {
+    if (lhs == null) {
+      return rhs == null;
+    }
     return lhs.equals(rhs);
   }
 
   @Override
   public int getUnsortedEncodedKeyComponentHashCode(Float key)
   {
-    return key.hashCode();
+    return key == null ? 0 : key.hashCode();
   }
 
   @Override

processing/src/main/java/io/druid/segment/LongDimensionIndexer.java

@@ -199,13 +199,16 @@ public int compareUnsortedEncodedKeyComponents(Long lhs, Long rhs)
   @Override
   public boolean checkUnsortedEncodedKeyComponentsEqual(Long lhs, Long rhs)
   {
+    if (lhs == null) {
+      return rhs == null;
+    }
     return lhs.equals(rhs);
   }
 
   @Override
   public int getUnsortedEncodedKeyComponentHashCode(Long key)
   {
-    return key.hashCode();
+    return key == null ? 0 : key.hashCode();
   }
 
   @Override

processing/src/test/java/io/druid/segment/incremental/IncrementalIndexTest.java

@@ -28,6 +28,8 @@
 import io.druid.data.input.Row;
 import io.druid.data.input.impl.DimensionSchema;
 import io.druid.data.input.impl.DimensionsSpec;
+import io.druid.data.input.impl.FloatDimensionSchema;
+import io.druid.data.input.impl.LongDimensionSchema;
 import io.druid.data.input.impl.StringDimensionSchema;
 import io.druid.java.util.common.ISE;
 import io.druid.java.util.common.granularity.Granularities;
@@ -76,8 +78,8 @@ public static Collection<?> constructorFeeder() throws IOException
     DimensionsSpec dimensions = new DimensionsSpec(
         Arrays.<DimensionSchema>asList(
             new StringDimensionSchema("string"),
-            new StringDimensionSchema("float"),
-            new StringDimensionSchema("long")
+            new FloatDimensionSchema("float"),
+            new LongDimensionSchema("long")
         ), null, null
     );
     AggregatorFactory[] metrics = {
@@ -206,6 +208,28 @@ public void controlTest() throws IndexSizeExceededException
 
   @Test
   public void testNullDimensionTransform() throws IndexSizeExceededException
+  {
+    IncrementalIndex<?> index = closer.closeLater(indexCreator.createIndex());
+    index.add(
+        new MapBasedInputRow(
+            new DateTime().minus(1).getMillis(),
+            Lists.newArrayList("string", "float", "long"),
+            ImmutableMap.<String, Object>of(
+                "string", Arrays.asList("A", null, "")
+            )
+        )
+    );
+
+    Row row = index.iterator().next();
+
+    Assert.assertEquals(Arrays.asList(new String[]{"", "", "A"}), row.getRaw("string"));
+    Assert.assertEquals(0.0f, row.getRaw("float"));
+    Assert.assertEquals(0L, row.getRaw("long"));
+  }
+
+
+  @Test
+  public void testUnparseableNumerics() throws IndexSizeExceededException
   {
     IncrementalIndex<?> index = closer.closeLater(indexCreator.createIndex());
     index.add(
@@ -214,17 +238,17 @@ public void testNullDimensionTransform() throws IndexSizeExceededException
             Lists.newArrayList("string", "float", "long"),
             ImmutableMap.<String, Object>of(
                 "string", Arrays.asList("A", null, ""),
-                "float", Arrays.asList(Float.MAX_VALUE, null, ""),
-                "long", Arrays.asList(Long.MIN_VALUE, null, "")
+                "float", "hello",
+                "long", "world"
             )
         )
     );
 
     Row row = index.iterator().next();
 
     Assert.assertEquals(Arrays.asList(new String[]{"", "", "A"}), row.getRaw("string"));
-    Assert.assertEquals(Arrays.asList(new String[]{"", "", String.valueOf(Float.MAX_VALUE)}), row.getRaw("float"));
-    Assert.assertEquals(Arrays.asList(new String[]{"", "", String.valueOf(Long.MIN_VALUE)}), row.getRaw("long"));
+    Assert.assertEquals(0.0f, row.getRaw("float"));
+    Assert.assertEquals(0L, row.getRaw("long"));
   }
 
   @Test