making [GreaterThan,LessThan,Equals] HavingSpecs more robust by carefully using long vs float for comparison

processing/src/main/java/io/druid/query/groupby/having/EqualToHavingSpec.java

@@ -62,9 +62,7 @@ public class EqualToHavingSpec implements HavingSpec
   @Override
   public boolean eval(Row row)
   {
-    float metricValue = row.getFloatMetric(aggregationName);
+    return HavingSpecMetricComparator.compare(row, aggregationName, value) == 0;
-
-    return Float.compare(value.floatValue(), metricValue) == 0;
   }
 
   @Override

processing/src/main/java/io/druid/query/groupby/having/GreaterThanHavingSpec.java

@@ -62,9 +62,7 @@ public class GreaterThanHavingSpec implements HavingSpec
   @Override
   public boolean eval(Row row)
   {
-    float metricValue = row.getFloatMetric(aggregationName);
+    return HavingSpecMetricComparator.compare(row, aggregationName, value) > 0;
-
-    return Float.compare(metricValue, value.floatValue()) > 0;
   }
 
   @Override

processing/src/main/java/io/druid/query/groupby/having/HavingSpecMetricComparator.java

@@ -0,0 +1,53 @@
+/*
+* Licensed to Metamarkets Group Inc. (Metamarkets) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. Metamarkets licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+
+package io.druid.query.groupby.having;
+
+import com.google.common.primitives.Floats;
+import com.google.common.primitives.Longs;
+import io.druid.data.input.Row;
+
+import java.util.regex.Pattern;
+
+/**
+ */
+class HavingSpecMetricComparator
+{
+  static final Pattern LONG_PAT = Pattern.compile("[-|+]?\\d+");
+
+  static int compare(Row row, String aggregationName, Number value)
+  {
+    Object metricValueObj = row.getRaw(aggregationName);
+    if (metricValueObj != null) {
+      if (metricValueObj instanceof Long) {
+        long l = ((Long) metricValueObj).longValue();
+        return Longs.compare(l, value.longValue());
+      } else if (metricValueObj instanceof String) {
+        String metricValueStr = (String) metricValueObj;
+        if (LONG_PAT.matcher(metricValueStr).matches()) {
+          long l = row.getLongMetric(aggregationName);
+          return Longs.compare(l, value.longValue());
+        }
+      }
+    }
+
+    float f = row.getFloatMetric(aggregationName);
+    return Floats.compare(f, value.floatValue());
+  }
+}

processing/src/main/java/io/druid/query/groupby/having/LessThanHavingSpec.java

@@ -60,9 +60,7 @@ public class LessThanHavingSpec implements HavingSpec
   @Override
   public boolean eval(Row row)
   {
-    float metricValue = row.getFloatMetric(aggregationName);
+    return HavingSpecMetricComparator.compare(row, aggregationName, value) < 0;
-
-    return Float.compare(metricValue, value.floatValue()) < 0;
   }
 
   @Override

processing/src/test/java/io/druid/query/groupby/having/HavingSpecMetricComparatorTest.java

@@ -0,0 +1,44 @@
+/*
+* Licensed to Metamarkets Group Inc. (Metamarkets) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. Metamarkets licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+
+package io.druid.query.groupby.having;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ */
+public class HavingSpecMetricComparatorTest
+{
+  @Test
+  public void testLongRegex()
+  {
+    Assert.assertTrue(HavingSpecMetricComparator.LONG_PAT.matcher("1").matches());
+    Assert.assertTrue(HavingSpecMetricComparator.LONG_PAT.matcher("12").matches());
+
+    Assert.assertFalse(HavingSpecMetricComparator.LONG_PAT.matcher("1.").matches());
+    Assert.assertFalse(HavingSpecMetricComparator.LONG_PAT.matcher("1.2").matches());
+    Assert.assertFalse(HavingSpecMetricComparator.LONG_PAT.matcher("1.23").matches());
+    Assert.assertFalse(HavingSpecMetricComparator.LONG_PAT.matcher("1E5").matches());
+    Assert.assertFalse(HavingSpecMetricComparator.LONG_PAT.matcher("1.23E5").matches());
+
+    Assert.assertFalse(HavingSpecMetricComparator.LONG_PAT.matcher("").matches());
+    Assert.assertFalse(HavingSpecMetricComparator.LONG_PAT.matcher("xyz").matches());
+  }
+}

processing/src/test/java/io/druid/query/groupby/having/HavingSpecTest.java

@@ -20,7 +20,6 @@ package io.druid.query.groupby.having;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Lists;
 import io.druid.data.input.MapBasedInputRow;
 import io.druid.data.input.Row;
 import io.druid.jackson.DefaultObjectMapper;
@@ -88,46 +87,54 @@ public class HavingSpecTest
 
   @Test
   public void testGreaterThanHavingSpec() {
-    GreaterThanHavingSpec spec = new GreaterThanHavingSpec("metric", 10.003);
+    GreaterThanHavingSpec spec = new GreaterThanHavingSpec("metric", Long.valueOf(Long.MAX_VALUE - 10));
-    assertFalse(spec.eval(ROW));
+    assertFalse(spec.eval(getTestRow(Long.valueOf(Long.MAX_VALUE - 10))));
+    assertFalse(spec.eval(getTestRow(Long.valueOf(Long.MAX_VALUE - 15))));
+    assertTrue(spec.eval(getTestRow(Long.valueOf(Long.MAX_VALUE - 5))));
+    assertTrue(spec.eval(getTestRow(String.valueOf(Long.MAX_VALUE - 5))));
+    assertFalse(spec.eval(getTestRow(100.05f)));
 
-    spec = new GreaterThanHavingSpec("metric", 10);
+    spec = new GreaterThanHavingSpec("metric", 100.56f);
-    assertFalse(spec.eval(ROW));
+    assertFalse(spec.eval(getTestRow(100.56f)));
-
+    assertFalse(spec.eval(getTestRow(90.53f)));
-    spec = new GreaterThanHavingSpec("metric", 9);
+    assertFalse(spec.eval(getTestRow("90.53f")));
-    assertTrue(spec.eval(ROW));
+    assertTrue(spec.eval(getTestRow(101.34f)));
-  }
+    assertTrue(spec.eval(getTestRow(Long.MAX_VALUE)));
-
-  private MapBasedInputRow makeRow(long ts, String dim, int value)
-  {
-    List<String> dimensions = Lists.newArrayList(dim);
-    Map<String, Object> metrics = ImmutableMap.of("metric", (Object) Float.valueOf(value));
-
-    return new MapBasedInputRow(ts, dimensions, metrics);
   }
 
   @Test
   public void testLessThanHavingSpec() {
-    LessThanHavingSpec spec = new LessThanHavingSpec("metric", 10);
+    LessThanHavingSpec spec = new LessThanHavingSpec("metric", Long.valueOf(Long.MAX_VALUE - 10));
-    assertFalse(spec.eval(ROW));
+    assertFalse(spec.eval(getTestRow(Long.valueOf(Long.MAX_VALUE - 10))));
+    assertTrue(spec.eval(getTestRow(Long.valueOf(Long.MAX_VALUE - 15))));
+    assertTrue(spec.eval(getTestRow(String.valueOf(Long.MAX_VALUE - 15))));
+    assertFalse(spec.eval(getTestRow(Long.valueOf(Long.MAX_VALUE - 5))));
+    assertTrue(spec.eval(getTestRow(100.05f)));
 
-    spec = new LessThanHavingSpec("metric", 11);
+    spec = new LessThanHavingSpec("metric", 100.56f);
-    assertTrue(spec.eval(ROW));
+    assertFalse(spec.eval(getTestRow(100.56f)));
+    assertTrue(spec.eval(getTestRow(90.53f)));
+    assertFalse(spec.eval(getTestRow(101.34f)));
+    assertFalse(spec.eval(getTestRow("101.34f")));
+    assertFalse(spec.eval(getTestRow(Long.MAX_VALUE)));
+  }
 
-    spec = new LessThanHavingSpec("metric", 9);
+  private Row getTestRow(Object metricValue)
-    assertFalse(spec.eval(ROW));
+  {
+    return new MapBasedInputRow(0, new ArrayList<String>(), ImmutableMap.of("metric", metricValue));
   }
 
   @Test
   public void testEqualHavingSpec() {
-    EqualToHavingSpec spec = new EqualToHavingSpec("metric", 10);
+    EqualToHavingSpec spec = new EqualToHavingSpec("metric", Long.valueOf(Long.MAX_VALUE - 10));
-    assertTrue(spec.eval(ROW));
+    assertTrue(spec.eval(getTestRow(Long.valueOf(Long.MAX_VALUE - 10))));
+    assertFalse(spec.eval(getTestRow(Long.valueOf(Long.MAX_VALUE - 5))));
+    assertFalse(spec.eval(getTestRow(100.05f)));
 
-    spec = new EqualToHavingSpec("metric", 9);
+    spec = new EqualToHavingSpec("metric", 100.56f);
-    assertFalse(spec.eval(ROW));
+    assertTrue(spec.eval(getTestRow(100.56f)));
-
+    assertFalse(spec.eval(getTestRow(90.53f)));
-    spec = new EqualToHavingSpec("metric", 11);
+    assertFalse(spec.eval(getTestRow(Long.MAX_VALUE)));
-    assertFalse(spec.eval(ROW));
   }
 
   private static class CountingHavingSpec implements HavingSpec {