Add default comparison to HavingSpecMetricComparator for custom Aggregator types (#6505)

* Add default comparison

* Switch to BigDecimal comparison

* Add comparator from AggFactory

* Fix indent

* Add tests

processing/src/main/java/org/apache/druid/query/groupby/having/HavingSpecMetricComparator.java

@@ -73,6 +73,14 @@ class HavingSpecMetricComparator
           double d = Double.parseDouble(metricValueStr);
           return Double.compare(d, value.doubleValue());
         }
+      } else if (aggregators != null && aggregators.containsKey(aggregationName)) {
+        // Use custom comparator in case of custom aggregation types
+        AggregatorFactory aggregatorFactory = aggregators.get(aggregationName);
+        return aggregatorFactory.getComparator()
+                                .compare(
+                                    aggregatorFactory.deserialize(metricValueObj),
+                                    aggregatorFactory.deserialize(value)
+                                );
       } else {
         throw new ISE("Unknown type of metric value: %s", metricValueObj);
       }

processing/src/test/java/org/apache/druid/query/TestBigDecimalSumAggregatorFactory.java

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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 org.apache.druid.query;
+
+import org.apache.druid.query.aggregation.DoubleSumAggregatorFactory;
+
+import javax.annotation.Nullable;
+import java.math.BigDecimal;
+import java.util.Comparator;
+
+public class TestBigDecimalSumAggregatorFactory extends DoubleSumAggregatorFactory
+{
+  public TestBigDecimalSumAggregatorFactory(String name, String fieldName)
+  {
+    super(name, fieldName);
+  }
+
+  @Override
+  @Nullable
+  public Object finalizeComputation(@Nullable Object object)
+  {
+    if (object instanceof Long) {
+      return BigDecimal.valueOf((Long) object);
+    } else if (object instanceof Double) {
+      return BigDecimal.valueOf((Double) object);
+    } else {
+      return object;
+    }
+  }
+
+  @Override
+  public Object deserialize(Object object)
+  {
+    if (object instanceof String) {
+      return BigDecimal.valueOf(Double.parseDouble((String) object));
+    } else if (object instanceof Double) {
+      return BigDecimal.valueOf((Double) object);
+    } else if (object instanceof Long) {
+      return BigDecimal.valueOf((Long) object);
+    }
+    return object;
+  }
+
+  @Override
+  public Comparator getComparator()
+  {
+    return new Comparator<BigDecimal>()
+    {
+      @Override
+      public int compare(BigDecimal o, BigDecimal o1)
+      {
+        return o.compareTo(o1);
+      }
+    };
+  }
+}

processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerTest.java

@@ -61,6 +61,7 @@ import org.apache.druid.query.QueryRunnerTestHelper;
 import org.apache.druid.query.QueryToolChest;
 import org.apache.druid.query.ResourceLimitExceededException;
 import org.apache.druid.query.Result;
+import org.apache.druid.query.TestBigDecimalSumAggregatorFactory;
 import org.apache.druid.query.aggregation.AggregatorFactory;
 import org.apache.druid.query.aggregation.CountAggregatorFactory;
 import org.apache.druid.query.aggregation.DoubleMaxAggregatorFactory;
@@ -3815,6 +3816,112 @@ public class GroupByQueryRunnerTest
     );
   }
 
+  @Test
+  public void testCustomAggregatorHavingSpec()
+  {
+    List<Row> expectedResults = Arrays.asList(
+        GroupByQueryRunnerTestHelper.createExpectedRow(
+            "2011-04-01",
+            "alias",
+            "automotive",
+            "rows",
+            1L,
+            "idxDouble",
+            135.885094d
+        ),
+        GroupByQueryRunnerTestHelper.createExpectedRow(
+            "2011-04-01",
+            "alias",
+            "entertainment",
+            "rows",
+            1L,
+            "idxDouble",
+            158.747224d
+        ),
+        GroupByQueryRunnerTestHelper.createExpectedRow(
+            "2011-04-01",
+            "alias",
+            "mezzanine",
+            "rows",
+            3L,
+            "idxDouble",
+            2871.8866900000003d
+        ),
+        GroupByQueryRunnerTestHelper.createExpectedRow(
+            "2011-04-01",
+            "alias",
+            "premium",
+            "rows",
+            3L,
+            "idxDouble",
+            2900.798647d
+        ),
+
+        GroupByQueryRunnerTestHelper.createExpectedRow(
+            "2011-04-02",
+            "alias",
+            "automotive",
+            "rows",
+            1L,
+            "idxDouble",
+            147.425935d
+        ),
+        GroupByQueryRunnerTestHelper.createExpectedRow(
+            "2011-04-02",
+            "alias",
+            "entertainment",
+            "rows",
+            1L,
+            "idxDouble",
+            166.016049d
+        ),
+        GroupByQueryRunnerTestHelper.createExpectedRow(
+            "2011-04-02",
+            "alias",
+            "mezzanine",
+            "rows",
+            3L,
+            "idxDouble",
+            2448.830613d
+        ),
+        GroupByQueryRunnerTestHelper.createExpectedRow(
+            "2011-04-02",
+            "alias",
+            "premium",
+            "rows",
+            3L,
+            "idxDouble",
+            2506.415148d
+        )
+    );
+
+    GroupByQuery query = GroupByQuery
+        .builder()
+        .setDataSource(QueryRunnerTestHelper.dataSource)
+        .setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
+        .setDimensions(new DefaultDimensionSpec("quality", "alias"))
+        .setAggregatorSpecs(
+            QueryRunnerTestHelper.rowsCount,
+            new TestBigDecimalSumAggregatorFactory("idxDouble", "index")
+        )
+        .setGranularity(QueryRunnerTestHelper.dayGran)
+        .setHavingSpec(
+            new OrHavingSpec(
+                ImmutableList.of(
+                    new EqualToHavingSpec("rows", 3L),
+                    new GreaterThanHavingSpec("idxDouble", 135.00d)
+                )
+            )
+        )
+        .build();
+
+    TestHelper.assertExpectedObjects(
+        expectedResults,
+        GroupByQueryRunnerTestHelper.runQuery(factory, runner, query),
+        ""
+    );
+  }
+
   @Test
   public void testGroupByWithRegEx()
   {