From d67633d1fd5e3354b6bba16abe0edd1eee0cbfcd Mon Sep 17 00:00:00 2001
From: Gian Merlino <gian@imply.io>
Date: Mon, 5 Feb 2018 22:12:54 -0800
Subject: [PATCH 1/4] 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).
---
 .../having/HavingSpecMetricComparator.java    | 27 +++++----
 .../query/groupby/having/HavingSpecTest.java  | 58 ++++++++++++++++++-
 2 files changed, 72 insertions(+), 13 deletions(-)

diff --git a/processing/src/main/java/io/druid/query/groupby/having/HavingSpecMetricComparator.java b/processing/src/main/java/io/druid/query/groupby/having/HavingSpecMetricComparator.java
index ff08202bf7cb..20349b26c110 100644
--- a/processing/src/main/java/io/druid/query/groupby/having/HavingSpecMetricComparator.java
+++ b/processing/src/main/java/io/druid/query/groupby/having/HavingSpecMetricComparator.java
@@ -24,9 +24,10 @@
 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 +47,22 @@ static int compare(Row row, String aggregationName, Number value, Map<String, Ag
         metricValueObj = aggregators.get(aggregationName).finalizeComputation(metricValueObj);
       }
 
-      if (metricValueObj instanceof Long) {
-        long l = ((Long) metricValueObj).longValue();
-        return Longs.compare(l, value.longValue());
+      if (metricValueObj instanceof Long || metricValueObj instanceof Integer) {
+        long n = ((Number) metricValueObj).longValue();
+
+        if (value instanceof Long || value instanceof Integer) {
+          return Longs.compare(n, value.longValue());
+        } else if (value instanceof Double || value instanceof Float) {
+          return BigDecimal.valueOf(n).compareTo(BigDecimal.valueOf(value.doubleValue()));
+        } else {
+          throw new ISE("Number was[%s]?!?", value.getClass().getName());
+        }
       } else if (metricValueObj instanceof Float) {
-        float l = ((Float) metricValueObj).floatValue();
-        return Floats.compare(l, value.floatValue());
+        float n = (Float) metricValueObj;
+        return Floats.compare(n, value.floatValue());
       } else if (metricValueObj instanceof Double) {
-        double l = ((Double) metricValueObj).longValue();
-        return Doubles.compare(l, value.doubleValue());
-      } else if (metricValueObj instanceof Integer) {
-        int l = ((Integer) metricValueObj).intValue();
-        return Ints.compare(l, value.intValue());
+        double n = (Double) metricValueObj;
+        return Doubles.compare(n, value.doubleValue());
       } else if (metricValueObj instanceof String) {
         String metricValueStr = (String) metricValueObj;
         if (LONG_PAT.matcher(metricValueStr).matches()) {
diff --git a/processing/src/test/java/io/druid/query/groupby/having/HavingSpecTest.java b/processing/src/test/java/io/druid/query/groupby/having/HavingSpecTest.java
index 18fc599e6bdd..d25cfae3ad8a 100644
--- a/processing/src/test/java/io/druid/query/groupby/having/HavingSpecTest.java
+++ b/processing/src/test/java/io/druid/query/groupby/having/HavingSpecTest.java
@@ -101,9 +101,8 @@ public void testTypeTypo()
         "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 void testEqualHavingSpec()
     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))); // True since 100.56f == (float) 100.56d
+    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.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

From 1d19a6fee90b5da3d057b45b5da3ee30724f9174 Mon Sep 17 00:00:00 2001
From: Gian Merlino <gian@imply.io>
Date: Tue, 6 Feb 2018 07:45:53 -0800
Subject: [PATCH 2/4] Add comments.

---
 .../query/groupby/having/HavingSpecMetricComparator.java      | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/processing/src/main/java/io/druid/query/groupby/having/HavingSpecMetricComparator.java b/processing/src/main/java/io/druid/query/groupby/having/HavingSpecMetricComparator.java
index 20349b26c110..70198b8f8abe 100644
--- a/processing/src/main/java/io/druid/query/groupby/having/HavingSpecMetricComparator.java
+++ b/processing/src/main/java/io/druid/query/groupby/having/HavingSpecMetricComparator.java
@@ -48,11 +48,15 @@ static int compare(Row row, String aggregationName, Number value, Map<String, Ag
       }
 
       if (metricValueObj instanceof Long || metricValueObj instanceof Integer) {
+        // Cast ints metrics to longs. It makes the code simpler and won't change the result.
         long n = ((Number) metricValueObj).longValue();
 
         if (value instanceof Long || value instanceof Integer) {
           return Longs.compare(n, value.longValue());
         } else if (value instanceof Double || value instanceof Float) {
+          // Use BigDecimal for comparison, a convenient way to handle edge cases without worrying about them
+          // ourselves. Cast the value to double since that's what BigDecimal wants. Should be fine since doubles can
+          // represent all float values.
           return BigDecimal.valueOf(n).compareTo(BigDecimal.valueOf(value.doubleValue()));
         } else {
           throw new ISE("Number was[%s]?!?", value.getClass().getName());

From 00a2a245d22323514753c54a3b00e24e945f4df1 Mon Sep 17 00:00:00 2001
From: Gian Merlino <gian@imply.io>
Date: Tue, 6 Feb 2018 07:48:22 -0800
Subject: [PATCH 3/4] Remove extraneous comment.

---
 .../test/java/io/druid/query/groupby/having/HavingSpecTest.java | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/processing/src/test/java/io/druid/query/groupby/having/HavingSpecTest.java b/processing/src/test/java/io/druid/query/groupby/having/HavingSpecTest.java
index d25cfae3ad8a..16417c5b7f9c 100644
--- a/processing/src/test/java/io/druid/query/groupby/having/HavingSpecTest.java
+++ b/processing/src/test/java/io/druid/query/groupby/having/HavingSpecTest.java
@@ -168,7 +168,7 @@ public void testEqualHavingSpec()
     assertFalse(spec.eval(getTestRow(100L)));
     assertFalse(spec.eval(getTestRow(100.0)));
     assertFalse(spec.eval(getTestRow(100d)));
-    assertTrue(spec.eval(getTestRow(100.56d))); // True since 100.56f == (float) 100.56d
+    assertTrue(spec.eval(getTestRow(100.56d)));
     assertFalse(spec.eval(getTestRow(90.53d)));
     assertTrue(spec.eval(getTestRow(100.56f)));
     assertFalse(spec.eval(getTestRow(90.53f)));

From 27677e05ed5a8116fba60c3a3cb770c246b3f243 Mon Sep 17 00:00:00 2001
From: Gian Merlino <gian@imply.io>
Date: Tue, 6 Feb 2018 10:56:17 -0800
Subject: [PATCH 4/4] Simplify code a bit.

---
 .../having/HavingSpecMetricComparator.java    | 34 ++++++++++++-------
 .../query/groupby/having/HavingSpecTest.java  |  2 +-
 2 files changed, 22 insertions(+), 14 deletions(-)

diff --git a/processing/src/main/java/io/druid/query/groupby/having/HavingSpecMetricComparator.java b/processing/src/main/java/io/druid/query/groupby/having/HavingSpecMetricComparator.java
index 70198b8f8abe..1b041dc1be5a 100644
--- a/processing/src/main/java/io/druid/query/groupby/having/HavingSpecMetricComparator.java
+++ b/processing/src/main/java/io/druid/query/groupby/having/HavingSpecMetricComparator.java
@@ -20,8 +20,6 @@
 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.java.util.common.ISE;
@@ -48,25 +46,28 @@ static int compare(Row row, String aggregationName, Number value, Map<String, Ag
       }
 
       if (metricValueObj instanceof Long || metricValueObj instanceof Integer) {
-        // Cast ints metrics to longs. It makes the code simpler and won't change the result.
+        // Cast int metrics to longs, it's fine since longs can represent all int values.
         long n = ((Number) metricValueObj).longValue();
 
         if (value instanceof Long || value instanceof Integer) {
           return Longs.compare(n, value.longValue());
         } else if (value instanceof Double || value instanceof Float) {
-          // Use BigDecimal for comparison, a convenient way to handle edge cases without worrying about them
-          // ourselves. Cast the value to double since that's what BigDecimal wants. Should be fine since doubles can
-          // represent all float values.
-          return BigDecimal.valueOf(n).compareTo(BigDecimal.valueOf(value.doubleValue()));
+          // Invert the comparison since we're providing n, value backwards.
+          return -compareDoubleToLong(value.doubleValue(), n);
+        } else {
+          throw new ISE("Number was[%s]?!?", value.getClass().getName());
+        }
+      } 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 Float) {
-        float n = (Float) metricValueObj;
-        return Floats.compare(n, value.floatValue());
-      } else if (metricValueObj instanceof Double) {
-        double n = (Double) metricValueObj;
-        return Doubles.compare(n, value.doubleValue());
       } else if (metricValueObj instanceof String) {
         String metricValueStr = (String) metricValueObj;
         if (LONG_PAT.matcher(metricValueStr).matches()) {
@@ -83,4 +84,11 @@ static int compare(Row row, String aggregationName, Number value, Map<String, Ag
       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));
+  }
 }
diff --git a/processing/src/test/java/io/druid/query/groupby/having/HavingSpecTest.java b/processing/src/test/java/io/druid/query/groupby/having/HavingSpecTest.java
index 16417c5b7f9c..183789a061b6 100644
--- a/processing/src/test/java/io/druid/query/groupby/having/HavingSpecTest.java
+++ b/processing/src/test/java/io/druid/query/groupby/having/HavingSpecTest.java
@@ -170,7 +170,7 @@ public void testEqualHavingSpec()
     assertFalse(spec.eval(getTestRow(100d)));
     assertTrue(spec.eval(getTestRow(100.56d)));
     assertFalse(spec.eval(getTestRow(90.53d)));
-    assertTrue(spec.eval(getTestRow(100.56f)));
+    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)));