Fix two improper casts in HavingSpecMetricComparator. (#5352)

* Fix two improper casts in HavingSpecMetricComparator.

Fixes two things:

1. An improper double-to-long cast when comparing double metrics to any
   kind of value, which was a regression from #4883.
2. An improper double-to-long cast when comparing a long/int metric to a
   double/float value: the value was cast to long/int, drawing strange
   conclusions like int 100 matching a havingSpec of equalTo(100.5).

* Add comments.

* Remove extraneous comment.

* Simplify code a bit.

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

@@ -20,13 +20,12 @@
 package io.druid.query.groupby.having;
 
 import com.google.common.primitives.Doubles;
-import com.google.common.primitives.Floats;
-import com.google.common.primitives.Ints;
 import com.google.common.primitives.Longs;
 import io.druid.data.input.Row;
-import io.druid.query.aggregation.AggregatorFactory;
 import io.druid.java.util.common.ISE;
+import io.druid.query.aggregation.AggregatorFactory;
 
+import java.math.BigDecimal;
 import java.util.Map;
 import java.util.regex.Pattern;
 
@@ -46,18 +45,29 @@ class HavingSpecMetricComparator
         metricValueObj = aggregators.get(aggregationName).finalizeComputation(metricValueObj);
       }
 
-      if (metricValueObj instanceof Long) {
+      if (metricValueObj instanceof Long || metricValueObj instanceof Integer) {
-        long l = ((Long) metricValueObj).longValue();
+        // Cast int metrics to longs, it's fine since longs can represent all int values.
-        return Longs.compare(l, value.longValue());
+        long n = ((Number) metricValueObj).longValue();
-      } else if (metricValueObj instanceof Float) {
+
-        float l = ((Float) metricValueObj).floatValue();
+        if (value instanceof Long || value instanceof Integer) {
-        return Floats.compare(l, value.floatValue());
+          return Longs.compare(n, value.longValue());
-      } else if (metricValueObj instanceof Double) {
+        } else if (value instanceof Double || value instanceof Float) {
-        double l = ((Double) metricValueObj).longValue();
+          // Invert the comparison since we're providing n, value backwards.
-        return Doubles.compare(l, value.doubleValue());
+          return -compareDoubleToLong(value.doubleValue(), n);
-      } else if (metricValueObj instanceof Integer) {
+        } else {
-        int l = ((Integer) metricValueObj).intValue();
+          throw new ISE("Number was[%s]?!?", value.getClass().getName());
-        return Ints.compare(l, value.intValue());
+        }
+      } else if (metricValueObj instanceof Double || metricValueObj instanceof Float) {
+        // Cast floats to doubles, it's fine since doubles can represent all float values.
+        double n = ((Number) metricValueObj).doubleValue();
+
+        if (value instanceof Long || value instanceof Integer) {
+          return compareDoubleToLong(n, value.longValue());
+        } else if (value instanceof Double || value instanceof Float) {
+          return Doubles.compare(n, value.doubleValue());
+        } else {
+          throw new ISE("Number was[%s]?!?", value.getClass().getName());
+        }
       } else if (metricValueObj instanceof String) {
         String metricValueStr = (String) metricValueObj;
         if (LONG_PAT.matcher(metricValueStr).matches()) {
@@ -74,4 +84,11 @@ class HavingSpecMetricComparator
       return Double.compare(0, value.doubleValue());
     }
   }
+
+  private static int compareDoubleToLong(final double a, final long b)
+  {
+    // Use BigDecimal when comparing integers vs floating points, a convenient way to handle all cases (like
+    // fractional values, values out of range of max long/max int) without worrying about them ourselves.
+    return BigDecimal.valueOf(a).compareTo(BigDecimal.valueOf(b));
+  }
 }

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

@@ -101,9 +101,8 @@ public class HavingSpecTest
         "value", 1.3
     );
     ObjectMapper mapper = new DefaultObjectMapper();
-    @SuppressWarnings("unused") // expected exception
+    // noinspection unused
     HavingSpec spec = mapper.convertValue(greaterMap, HavingSpec.class);
-
   }
 
   @Test
@@ -156,9 +155,64 @@ public class HavingSpecTest
     assertFalse(spec.eval(getTestRow(100.05f)));
 
     spec = new EqualToHavingSpec("metric", 100.56f);
+    assertFalse(spec.eval(getTestRow(100L)));
+    assertFalse(spec.eval(getTestRow(100.0)));
+    assertFalse(spec.eval(getTestRow(100d)));
+    assertFalse(spec.eval(getTestRow(100.56d))); // False since 100.56d != (double) 100.56f
+    assertFalse(spec.eval(getTestRow(90.53d)));
     assertTrue(spec.eval(getTestRow(100.56f)));
     assertFalse(spec.eval(getTestRow(90.53f)));
     assertFalse(spec.eval(getTestRow(Long.MAX_VALUE)));
+
+    spec = new EqualToHavingSpec("metric", 100.56d);
+    assertFalse(spec.eval(getTestRow(100L)));
+    assertFalse(spec.eval(getTestRow(100.0)));
+    assertFalse(spec.eval(getTestRow(100d)));
+    assertTrue(spec.eval(getTestRow(100.56d)));
+    assertFalse(spec.eval(getTestRow(90.53d)));
+    assertFalse(spec.eval(getTestRow(100.56f))); // False since 100.56d != (double) 100.56f
+    assertFalse(spec.eval(getTestRow(90.53f)));
+    assertFalse(spec.eval(getTestRow(Long.MAX_VALUE)));
+
+    spec = new EqualToHavingSpec("metric", 100.0f);
+    assertTrue(spec.eval(getTestRow(100L)));
+    assertTrue(spec.eval(getTestRow(100.0)));
+    assertTrue(spec.eval(getTestRow(100d)));
+    assertFalse(spec.eval(getTestRow(100.56d)));
+    assertFalse(spec.eval(getTestRow(90.53d)));
+    assertFalse(spec.eval(getTestRow(100.56f)));
+    assertFalse(spec.eval(getTestRow(90.53f)));
+    assertFalse(spec.eval(getTestRow(Long.MAX_VALUE)));
+
+    spec = new EqualToHavingSpec("metric", 100.0d);
+    assertTrue(spec.eval(getTestRow(100L)));
+    assertTrue(spec.eval(getTestRow(100.0)));
+    assertTrue(spec.eval(getTestRow(100d)));
+    assertFalse(spec.eval(getTestRow(100.56d)));
+    assertFalse(spec.eval(getTestRow(90.53d)));
+    assertFalse(spec.eval(getTestRow(100.56f)));
+    assertFalse(spec.eval(getTestRow(90.53f)));
+    assertFalse(spec.eval(getTestRow(Long.MAX_VALUE)));
+
+    spec = new EqualToHavingSpec("metric", 100);
+    assertTrue(spec.eval(getTestRow(100L)));
+    assertTrue(spec.eval(getTestRow(100.0)));
+    assertTrue(spec.eval(getTestRow(100d)));
+    assertFalse(spec.eval(getTestRow(100.56d)));
+    assertFalse(spec.eval(getTestRow(90.53d)));
+    assertFalse(spec.eval(getTestRow(100.56f)));
+    assertFalse(spec.eval(getTestRow(90.53f)));
+    assertFalse(spec.eval(getTestRow(Long.MAX_VALUE)));
+
+    spec = new EqualToHavingSpec("metric", 100L);
+    assertTrue(spec.eval(getTestRow(100L)));
+    assertTrue(spec.eval(getTestRow(100.0)));
+    assertTrue(spec.eval(getTestRow(100d)));
+    assertFalse(spec.eval(getTestRow(100.56d)));
+    assertFalse(spec.eval(getTestRow(90.53d)));
+    assertFalse(spec.eval(getTestRow(100.56f)));
+    assertFalse(spec.eval(getTestRow(90.53f)));
+    assertFalse(spec.eval(getTestRow(Long.MAX_VALUE)));
   }
 
   private static class CountingHavingSpec extends BaseHavingSpec