GREATEST/LEAST post-aggregators in SQL (#8719)

* implement shell for greatest sql aggregator with hardcoded long values

* implement functional long greatest aggregator for direct access columns

* implement greatest & least sql aggregators for long & double types using abstract base class

* add javadocs, unit tests & handling for floats for greatest/least postaggregations

* minor checkstyle fix

* improve naming for the test cases

* make inner class static

* remove blank lines to retest travis build

* change trivial text to rerun travis build

* implement suggested updates for greatest/least sql aggs & fix checkstyle issues

* fix stale comments in greatest/least sql aggs abstract base

* Update sql.md

* improve sql function definitions for greatest/least sql aggs

* add more tests for greatest/least sql aggs

* add tests to cover invalid greatest/least sql expressions

* rename & reorder greatest least sql tests

docs/querying/sql.md

@@ -180,6 +180,8 @@ Only the COUNT aggregation can accept DISTINCT.
 |`SUM(expr)`|Sums numbers.|
 |`MIN(expr)`|Takes the minimum of numbers.|
 |`MAX(expr)`|Takes the maximum of numbers.|
+|`LEAST(expr1, [expr2, ...])`|Takes the minimum of numbers across one or more expression(s).|
+|`GREATEST(expr1, [expr2, ...])`|Takes the maximum of numbers across one or more expression(s).|
 |`AVG(expr)`|Averages numbers.|
 |`APPROX_COUNT_DISTINCT(expr)`|Counts distinct values of expr, which can be a regular column or a hyperUnique column. This is always approximate, regardless of the value of "useApproximateCountDistinct". This uses Druid's built-in "cardinality" or "hyperUnique" aggregators. See also `COUNT(DISTINCT expr)`.|
 |`APPROX_COUNT_DISTINCT_DS_HLL(expr, [lgK, tgtHllType])`|Counts distinct values of expr, which can be a regular column or an [HLL sketch](../development/extensions-core/datasketches-hll.html) column. The `lgK` and `tgtHllType` parameters are described in the HLL sketch documentation. This is always approximate, regardless of the value of "useApproximateCountDistinct". See also `COUNT(DISTINCT expr)`. The [DataSketches extension](../development/extensions-core/datasketches-extension.html) must be loaded to use this function.|

sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/GreatestSqlAggregator.java

@@ -0,0 +1,136 @@
+/*
+ * 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.sql.calcite.aggregation.builtin;
+
+import org.apache.calcite.sql.SqlAggFunction;
+import org.apache.calcite.sql.SqlFunctionCategory;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.type.InferTypes;
+import org.apache.calcite.sql.type.OperandTypes;
+import org.apache.calcite.sql.type.ReturnTypes;
+import org.apache.calcite.sql.type.SqlTypeTransforms;
+import org.apache.calcite.util.Optionality;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.math.expr.ExprMacroTable;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.query.aggregation.DoubleMaxAggregatorFactory;
+import org.apache.druid.query.aggregation.LongMaxAggregatorFactory;
+import org.apache.druid.query.aggregation.PostAggregator;
+import org.apache.druid.query.aggregation.post.DoubleGreatestPostAggregator;
+import org.apache.druid.query.aggregation.post.LongGreatestPostAggregator;
+import org.apache.druid.segment.column.ValueType;
+
+import java.util.List;
+
+/**
+ * Calcite integration class for Greatest post aggregators of Long & Double types.
+ * It applies Max aggregators over the provided fields/expressions & combines their results via Field access post aggregators.
+ */
+public class GreatestSqlAggregator extends MultiColumnSqlAggregator
+{
+  private static final SqlAggFunction FUNCTION_INSTANCE = new GreatestSqlAggFunction();
+  private static final String NAME = "GREATEST";
+
+  @Override
+  public SqlAggFunction calciteFunction()
+  {
+    return FUNCTION_INSTANCE;
+  }
+
+  @Override
+  AggregatorFactory createAggregatorFactory(
+      ValueType valueType,
+      String prefixedName,
+      FieldInfo fieldInfo,
+      ExprMacroTable macroTable
+  )
+  {
+    final AggregatorFactory aggregatorFactory;
+    switch (valueType) {
+      case LONG:
+        aggregatorFactory = new LongMaxAggregatorFactory(prefixedName, fieldInfo.fieldName, fieldInfo.expression, macroTable);
+        break;
+      case FLOAT:
+      case DOUBLE:
+        aggregatorFactory = new DoubleMaxAggregatorFactory(prefixedName, fieldInfo.fieldName, fieldInfo.expression, macroTable);
+        break;
+      default:
+        throw new ISE("Cannot create aggregator factory for type[%s]", valueType);
+    }
+    return aggregatorFactory;
+  }
+
+  @Override
+  PostAggregator createFinalPostAggregator(
+      ValueType valueType,
+      String name,
+      List<PostAggregator> postAggregators
+  )
+  {
+    final PostAggregator finalPostAggregator;
+    switch (valueType) {
+      case LONG:
+        finalPostAggregator = new LongGreatestPostAggregator(name, postAggregators);
+        break;
+      case FLOAT:
+      case DOUBLE:
+        finalPostAggregator = new DoubleGreatestPostAggregator(name, postAggregators);
+        break;
+      default:
+        throw new ISE("Cannot create aggregator factory for type[%s]", valueType);
+    }
+    return finalPostAggregator;
+  }
+
+  /**
+   * Calcite SQL function definition
+   */
+  private static class GreatestSqlAggFunction extends SqlAggFunction
+  {
+    GreatestSqlAggFunction()
+    {
+      /*
+       * The constructor params are explained as follows,
+       * name: SQL function name
+       * sqlIdentifier: null for built-in functions
+       * kind: SqlKind.GREATEST
+       * returnTypeInference: biggest operand type & nullable if any of the operands is nullable
+       * operandTypeInference: same as return type
+       * operandTypeChecker: variadic function with at least one argument
+       * funcType: System
+       * requiresOrder: No
+       * requiresOver: No
+       * requiresGroupOrder: Not allowed
+       */
+      super(
+          NAME,
+          null,
+          SqlKind.GREATEST,
+          ReturnTypes.cascade(ReturnTypes.LEAST_RESTRICTIVE, SqlTypeTransforms.TO_NULLABLE),
+          InferTypes.RETURN_TYPE,
+          OperandTypes.ONE_OR_MORE,
+          SqlFunctionCategory.SYSTEM,
+          false,
+          false,
+          Optionality.FORBIDDEN
+      );
+    }
+  }
+}

sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/LeastSqlAggregator.java

@@ -0,0 +1,136 @@
+/*
+ * 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.sql.calcite.aggregation.builtin;
+
+import org.apache.calcite.sql.SqlAggFunction;
+import org.apache.calcite.sql.SqlFunctionCategory;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.type.InferTypes;
+import org.apache.calcite.sql.type.OperandTypes;
+import org.apache.calcite.sql.type.ReturnTypes;
+import org.apache.calcite.sql.type.SqlTypeTransforms;
+import org.apache.calcite.util.Optionality;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.math.expr.ExprMacroTable;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.query.aggregation.DoubleMinAggregatorFactory;
+import org.apache.druid.query.aggregation.LongMinAggregatorFactory;
+import org.apache.druid.query.aggregation.PostAggregator;
+import org.apache.druid.query.aggregation.post.DoubleLeastPostAggregator;
+import org.apache.druid.query.aggregation.post.LongLeastPostAggregator;
+import org.apache.druid.segment.column.ValueType;
+
+import java.util.List;
+
+/**
+ * Calcite integration class for Least post aggregators of Long & Double types.
+ * It applies Min aggregators over the provided fields/expressions & combines their results via Field access post aggregators.
+ */
+public class LeastSqlAggregator extends MultiColumnSqlAggregator
+{
+  private static final SqlAggFunction FUNCTION_INSTANCE = new LeastSqlAggFunction();
+  private static final String NAME = "LEAST";
+
+  @Override
+  public SqlAggFunction calciteFunction()
+  {
+    return FUNCTION_INSTANCE;
+  }
+
+  @Override
+  AggregatorFactory createAggregatorFactory(
+      ValueType valueType,
+      String prefixedName,
+      FieldInfo fieldInfo,
+      ExprMacroTable macroTable
+  )
+  {
+    final AggregatorFactory aggregatorFactory;
+    switch (valueType) {
+      case LONG:
+        aggregatorFactory = new LongMinAggregatorFactory(prefixedName, fieldInfo.fieldName, fieldInfo.expression, macroTable);
+        break;
+      case FLOAT:
+      case DOUBLE:
+        aggregatorFactory = new DoubleMinAggregatorFactory(prefixedName, fieldInfo.fieldName, fieldInfo.expression, macroTable);
+        break;
+      default:
+        throw new ISE("Cannot create aggregator factory for type[%s]", valueType);
+    }
+    return aggregatorFactory;
+  }
+
+  @Override
+  PostAggregator createFinalPostAggregator(
+      ValueType valueType,
+      String name,
+      List<PostAggregator> postAggregators
+  )
+  {
+    final PostAggregator finalPostAggregator;
+    switch (valueType) {
+      case LONG:
+        finalPostAggregator = new LongLeastPostAggregator(name, postAggregators);
+        break;
+      case FLOAT:
+      case DOUBLE:
+        finalPostAggregator = new DoubleLeastPostAggregator(name, postAggregators);
+        break;
+      default:
+        throw new ISE("Cannot create aggregator factory for type[%s]", valueType);
+    }
+    return finalPostAggregator;
+  }
+
+  /**
+   * Calcite SQL function definition
+   */
+  private static class LeastSqlAggFunction extends SqlAggFunction
+  {
+    LeastSqlAggFunction()
+    {
+      /*
+       * The constructor params are explained as follows,
+       * name: SQL function name
+       * sqlIdentifier: null for built-in functions
+       * kind: SqlKind.LEAST
+       * returnTypeInference: biggest operand type & nullable if any of the operands is nullable
+       * operandTypeInference: same as return type
+       * operandTypeChecker: variadic function with at least one argument
+       * funcType: System
+       * requiresOrder: No
+       * requiresOver: No
+       * requiresGroupOrder: Not allowed
+       */
+      super(
+          NAME,
+          null,
+          SqlKind.LEAST,
+          ReturnTypes.cascade(ReturnTypes.LEAST_RESTRICTIVE, SqlTypeTransforms.TO_NULLABLE),
+          InferTypes.RETURN_TYPE,
+          OperandTypes.ONE_OR_MORE,
+          SqlFunctionCategory.SYSTEM,
+          false,
+          false,
+          Optionality.FORBIDDEN
+      );
+    }
+  }
+}

sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/MultiColumnSqlAggregator.java

@@ -0,0 +1,152 @@
+/*
+ * 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.sql.calcite.aggregation.builtin;
+
+import com.google.common.base.Preconditions;
+import org.apache.calcite.rel.core.AggregateCall;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.druid.math.expr.ExprMacroTable;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.query.aggregation.PostAggregator;
+import org.apache.druid.query.aggregation.post.FieldAccessPostAggregator;
+import org.apache.druid.segment.column.ValueType;
+import org.apache.druid.sql.calcite.aggregation.Aggregation;
+import org.apache.druid.sql.calcite.aggregation.Aggregations;
+import org.apache.druid.sql.calcite.aggregation.SqlAggregator;
+import org.apache.druid.sql.calcite.expression.DruidExpression;
+import org.apache.druid.sql.calcite.planner.Calcites;
+import org.apache.druid.sql.calcite.planner.PlannerContext;
+import org.apache.druid.sql.calcite.rel.VirtualColumnRegistry;
+import org.apache.druid.sql.calcite.table.RowSignature;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Abstraction for simple multi-column post aggregators like greatest, least
+ */
+public abstract class MultiColumnSqlAggregator implements SqlAggregator
+{
+  /**
+   * Useful Abstraction for passing field information to subclasses from shared parent methods
+   */
+  protected static class FieldInfo
+  {
+    final String fieldName;
+    final String expression;
+
+    private FieldInfo(String fieldName, String expression)
+    {
+      this.fieldName = fieldName;
+      this.expression = expression;
+    }
+
+    public static FieldInfo fromFieldName(String fieldName)
+    {
+      return new FieldInfo(fieldName, null);
+    }
+
+    public static FieldInfo fromExpression(String expression)
+    {
+      return new FieldInfo(null, expression);
+    }
+  }
+
+  @Nullable
+  @Override
+  public Aggregation toDruidAggregation(
+      final PlannerContext plannerContext,
+      final RowSignature rowSignature,
+      final VirtualColumnRegistry virtualColumnRegistry,
+      final RexBuilder rexBuilder,
+      final String name,
+      final AggregateCall aggregateCall,
+      final Project project,
+      final List<Aggregation> existingAggregations,
+      final boolean finalizeAggregations
+  )
+  {
+    if (aggregateCall.isDistinct()) {
+      return null;
+    }
+
+    final List<DruidExpression> arguments = Aggregations.getArgumentsForSimpleAggregator(
+        plannerContext,
+        rowSignature,
+        aggregateCall,
+        project
+    );
+
+    if (arguments == null) {
+      return null;
+    }
+
+    final ExprMacroTable macroTable = plannerContext.getExprMacroTable();
+    final List<FieldInfo> fieldInfoList = new ArrayList<>();
+
+    // Convert arguments to concise field information
+    for (DruidExpression argument : arguments) {
+      if (argument.isDirectColumnAccess()) {
+        fieldInfoList.add(FieldInfo.fromFieldName(argument.getDirectColumn()));
+      } else {
+        fieldInfoList.add(FieldInfo.fromExpression(argument.getExpression()));
+      }
+    }
+    Preconditions.checkArgument(!fieldInfoList.isEmpty(), "FieldInfoList should not be empty");
+    return getAggregation(name, aggregateCall, macroTable, fieldInfoList);
+  }
+
+  private Aggregation getAggregation(
+      String name,
+      AggregateCall aggregateCall,
+      ExprMacroTable macroTable,
+      List<FieldInfo> fieldInfoList
+  )
+  {
+    final ValueType valueType = Calcites.getValueTypeForSqlTypeName(aggregateCall.getType().getSqlTypeName());
+    List<AggregatorFactory> aggregatorFactories = new ArrayList<>();
+    List<PostAggregator> postAggregators = new ArrayList<>();
+
+    // Delegate aggregator factory construction to subclasses for provided fields.
+    // Create corresponding field access post aggregators.
+    int id = 0;
+    for (FieldInfo fieldInfo : fieldInfoList) {
+      String prefixedName = Calcites.makePrefixedName(name, String.valueOf(id++));
+      postAggregators.add(new FieldAccessPostAggregator(null, prefixedName));
+      aggregatorFactories.add(createAggregatorFactory(valueType, prefixedName, fieldInfo, macroTable));
+    }
+    // Delegate final post aggregator construction to subclasses by passing the above aggregators.
+    final PostAggregator finalPostAggregator = createFinalPostAggregator(valueType, name, postAggregators);
+    return Aggregation.create(aggregatorFactories, finalPostAggregator);
+  }
+
+  abstract AggregatorFactory createAggregatorFactory(
+      ValueType valueType,
+      String prefixedName,
+      FieldInfo fieldInfo,
+      ExprMacroTable macroTable);
+
+  abstract PostAggregator createFinalPostAggregator(
+      ValueType valueType,
+      String name,
+      List<PostAggregator> postAggregators);
+}

sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidOperatorTable.java

@@ -37,6 +37,8 @@ import org.apache.druid.sql.calcite.aggregation.builtin.ApproxCountDistinctSqlAg
 import org.apache.druid.sql.calcite.aggregation.builtin.AvgSqlAggregator;
 import org.apache.druid.sql.calcite.aggregation.builtin.CountSqlAggregator;
 import org.apache.druid.sql.calcite.aggregation.builtin.EarliestLatestAnySqlAggregator;
+import org.apache.druid.sql.calcite.aggregation.builtin.GreatestSqlAggregator;
+import org.apache.druid.sql.calcite.aggregation.builtin.LeastSqlAggregator;
 import org.apache.druid.sql.calcite.aggregation.builtin.MaxSqlAggregator;
 import org.apache.druid.sql.calcite.aggregation.builtin.MinSqlAggregator;
 import org.apache.druid.sql.calcite.aggregation.builtin.SumSqlAggregator;
@@ -124,6 +126,8 @@ public class DruidOperatorTable implements SqlOperatorTable
           .add(EarliestLatestAnySqlAggregator.ANY_VALUE)
           .add(new MinSqlAggregator())
           .add(new MaxSqlAggregator())
+          .add(new GreatestSqlAggregator())
+          .add(new LeastSqlAggregator())
           .add(new SumSqlAggregator())
           .add(new SumZeroSqlAggregator())
           .build();

sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java

@@ -26,6 +26,7 @@ import org.apache.calcite.tools.ValidationException;
 import org.apache.druid.common.config.NullHandling;
 import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.Intervals;
 import org.apache.druid.java.util.common.JodaUtils;
 import org.apache.druid.java.util.common.granularity.Granularities;
@@ -38,6 +39,7 @@ import org.apache.druid.query.ResourceLimitExceededException;
 import org.apache.druid.query.TableDataSource;
 import org.apache.druid.query.aggregation.CountAggregatorFactory;
 import org.apache.druid.query.aggregation.DoubleMaxAggregatorFactory;
+import org.apache.druid.query.aggregation.DoubleMinAggregatorFactory;
 import org.apache.druid.query.aggregation.DoubleSumAggregatorFactory;
 import org.apache.druid.query.aggregation.FilteredAggregatorFactory;
 import org.apache.druid.query.aggregation.FloatMaxAggregatorFactory;
@@ -61,8 +63,12 @@ import org.apache.druid.query.aggregation.last.FloatLastAggregatorFactory;
 import org.apache.druid.query.aggregation.last.LongLastAggregatorFactory;
 import org.apache.druid.query.aggregation.last.StringLastAggregatorFactory;
 import org.apache.druid.query.aggregation.post.ArithmeticPostAggregator;
+import org.apache.druid.query.aggregation.post.DoubleGreatestPostAggregator;
+import org.apache.druid.query.aggregation.post.DoubleLeastPostAggregator;
 import org.apache.druid.query.aggregation.post.FieldAccessPostAggregator;
 import org.apache.druid.query.aggregation.post.FinalizingFieldAccessPostAggregator;
+import org.apache.druid.query.aggregation.post.LongGreatestPostAggregator;
+import org.apache.druid.query.aggregation.post.LongLeastPostAggregator;
 import org.apache.druid.query.dimension.DefaultDimensionSpec;
 import org.apache.druid.query.dimension.ExtractionDimensionSpec;
 import org.apache.druid.query.extraction.RegexDimExtractionFn;
@@ -6017,6 +6023,444 @@ public class CalciteQueryTest extends BaseCalciteQueryTest
     );
   }
 
+  @Test
+  public void testGreatestLongAndDoubleWithGroupBy() throws Exception
+  {
+    // Cannot vectorize due to virtual columns.
+    cannotVectorize();
+
+    testQuery(
+        "SELECT * FROM ("
+        + "  SELECT greatest(cntl1, cntl2), greatest(cntd1, cntd2) FROM (\n"
+        + "      SELECT TIME_FLOOR(__time, 'P1D') AS t,\n"
+        + "          count(1) AS cntl1, 10 AS cntl2,\n"
+        + "          (1.2 + count(1)) AS cntd1, 10.2 AS cntd2\n"
+        + "      FROM \"foo\"\n"
+        + "      GROUP BY 1\n"
+        + "  )"
+        + ")\n",
+        ImmutableList.of(
+            GroupByQuery.builder()
+                        .setDataSource(
+                            new QueryDataSource(
+                                GroupByQuery.builder()
+                                            .setDataSource(CalciteTests.DATASOURCE1)
+                                            .setInterval(querySegmentSpec(Filtration.eternity()))
+                                            .setGranularity(Granularities.ALL)
+                                            .setVirtualColumns(
+                                                expressionVirtualColumn(
+                                                    "v0",
+                                                    "timestamp_floor(\"__time\",'P1D',null,'UTC')",
+                                                    ValueType.LONG
+                                                )
+                                            )
+                                            .setDimensions(dimensions(new DefaultDimensionSpec(
+                                                "v0",
+                                                "v0",
+                                                ValueType.LONG
+                                            )))
+                                            .setAggregatorSpecs(aggregators(new CountAggregatorFactory("a0")))
+                                            .setContext(QUERY_CONTEXT_DEFAULT)
+                                            .build()
+                            )
+                        )
+                        .setInterval(querySegmentSpec(Filtration.eternity()))
+                        .setGranularity(Granularities.ALL)
+                        .setAggregatorSpecs(aggregators(
+                            new LongMaxAggregatorFactory("_a0:0", "a0"),
+                            new LongMaxAggregatorFactory("_a0:1", null, "10", ExprMacroTable.nil()),
+                            new DoubleMaxAggregatorFactory("_a1:0", null, "(1.2 + \"a0\")", ExprMacroTable.nil()),
+                            new DoubleMaxAggregatorFactory("_a1:1", null, "10.2", ExprMacroTable.nil())
+                        ))
+                        .setPostAggregatorSpecs(
+                            ImmutableList.of(
+                                new LongGreatestPostAggregator(
+                                    "_a0",
+                                    ImmutableList.of(
+                                        new FieldAccessPostAggregator(null, "_a0:0"),
+                                        new FieldAccessPostAggregator(null, "_a0:1")
+                                    )
+                                ),
+                                new DoubleGreatestPostAggregator(
+                                    "_a1",
+                                    ImmutableList.of(
+                                        new FieldAccessPostAggregator(null, "_a1:0"),
+                                        new FieldAccessPostAggregator(null, "_a1:1")
+                                    )
+                                )
+                            )
+                        )
+                        .setContext(QUERY_CONTEXT_DEFAULT)
+                        .build()
+        ),
+        ImmutableList.of(new Object[]{10L, 10.2D})
+    );
+  }
+
+  @Test
+  public void testLeastLongAndDoubleWithGroupBy() throws Exception
+  {
+    // Cannot vectorize due to virtual columns.
+    cannotVectorize();
+
+    testQuery(
+        "SELECT * FROM ("
+        + "  SELECT least(cntl1, cntl2), least(cntd1, cntd2) FROM (\n"
+        + "      SELECT TIME_FLOOR(__time, 'P1D') AS t,\n"
+        + "          count(1) AS cntl1, 10 AS cntl2,\n"
+        + "          (1.2 + count(1)) AS cntd1, 10.2 AS cntd2\n"
+        + "      FROM \"foo\"\n"
+        + "      GROUP BY 1\n"
+        + "  )"
+        + ")\n",
+        ImmutableList.of(
+            GroupByQuery.builder()
+                        .setDataSource(
+                            new QueryDataSource(
+                                GroupByQuery.builder()
+                                            .setDataSource(CalciteTests.DATASOURCE1)
+                                            .setInterval(querySegmentSpec(Filtration.eternity()))
+                                            .setGranularity(Granularities.ALL)
+                                            .setVirtualColumns(
+                                                expressionVirtualColumn(
+                                                    "v0",
+                                                    "timestamp_floor(\"__time\",'P1D',null,'UTC')",
+                                                    ValueType.LONG
+                                                )
+                                            )
+                                            .setDimensions(dimensions(new DefaultDimensionSpec(
+                                                "v0",
+                                                "v0",
+                                                ValueType.LONG
+                                            )))
+                                            .setAggregatorSpecs(aggregators(new CountAggregatorFactory("a0")))
+                                            .setContext(QUERY_CONTEXT_DEFAULT)
+                                            .build()
+                            )
+                        )
+                        .setInterval(querySegmentSpec(Filtration.eternity()))
+                        .setGranularity(Granularities.ALL)
+                        .setAggregatorSpecs(aggregators(
+                            new LongMinAggregatorFactory("_a0:0", "a0"),
+                            new LongMinAggregatorFactory("_a0:1", null, "10", ExprMacroTable.nil()),
+                            new DoubleMinAggregatorFactory("_a1:0", null, "(1.2 + \"a0\")", ExprMacroTable.nil()),
+                            new DoubleMinAggregatorFactory("_a1:1", null, "10.2", ExprMacroTable.nil())
+                        ))
+                        .setPostAggregatorSpecs(
+                            ImmutableList.of(
+                                new LongLeastPostAggregator(
+                                    "_a0",
+                                    ImmutableList.of(
+                                        new FieldAccessPostAggregator(null, "_a0:0"),
+                                        new FieldAccessPostAggregator(null, "_a0:1")
+                                    )
+                                ),
+                                new DoubleLeastPostAggregator(
+                                    "_a1",
+                                    ImmutableList.of(
+                                        new FieldAccessPostAggregator(null, "_a1:0"),
+                                        new FieldAccessPostAggregator(null, "_a1:1")
+                                    )
+                                )
+                            )
+                        )
+                        .setContext(QUERY_CONTEXT_DEFAULT)
+                        .build()
+        ),
+        ImmutableList.of(new Object[]{1L, 2.2D})
+    );
+  }
+
+  @Test
+  public void testGreatestSingleColumnPostAggregations() throws Exception
+  {
+    // Cannot vectorize due to virtual columns.
+    cannotVectorize();
+
+    testQuery(
+        "SELECT\n"
+        + "      greatest(cnt), greatest(m1), greatest(m2)\n"
+        + "      FROM \"foo\"\n",
+        ImmutableList.of(
+            Druids.newTimeseriesQueryBuilder()
+                  .dataSource(CalciteTests.DATASOURCE1)
+                  .intervals(querySegmentSpec(Filtration.eternity()))
+                  .granularity(Granularities.ALL)
+                  .aggregators(aggregators(
+                      new LongMaxAggregatorFactory("a0:0", "cnt"),
+                      new DoubleMaxAggregatorFactory("a1:0", "m1"),
+                      new DoubleMaxAggregatorFactory("a2:0", "m2")
+                  ))
+                  .postAggregators(ImmutableList.of(
+                      new LongGreatestPostAggregator(
+                          "a0",
+                          ImmutableList.of(
+                              new FieldAccessPostAggregator(null, "a0:0")
+                          )
+                      ),
+                      new DoubleGreatestPostAggregator(
+                          "a1",
+                          ImmutableList.of(
+                              new FieldAccessPostAggregator(null, "a1:0")
+                          )
+                      ),
+                      new DoubleGreatestPostAggregator(
+                          "a2",
+                          ImmutableList.of(
+                              new FieldAccessPostAggregator(null, "a2:0")
+                          )
+                      )
+                                   )
+                  )
+                  .context(TIMESERIES_CONTEXT_DEFAULT)
+                  .build()
+        ),
+        ImmutableList.of(new Object[]{1L, 6.0F, 6.0D})
+    );
+  }
+
+  @Test
+  public void testLeastSingleColumnPostAggregations() throws Exception
+  {
+    // Cannot vectorize due to virtual columns.
+    cannotVectorize();
+
+    testQuery(
+        "SELECT\n"
+        + "      least(cnt), least(m1), least(m2)\n"
+        + "      FROM \"foo\"\n",
+        ImmutableList.of(
+            Druids.newTimeseriesQueryBuilder()
+                  .dataSource(CalciteTests.DATASOURCE1)
+                  .intervals(querySegmentSpec(Filtration.eternity()))
+                  .granularity(Granularities.ALL)
+                  .aggregators(aggregators(
+                      new LongMinAggregatorFactory("a0:0", "cnt"),
+                      new DoubleMinAggregatorFactory("a1:0", "m1"),
+                      new DoubleMinAggregatorFactory("a2:0", "m2")
+                  ))
+                  .postAggregators(ImmutableList.of(
+                      new LongLeastPostAggregator(
+                          "a0",
+                          ImmutableList.of(
+                              new FieldAccessPostAggregator(null, "a0:0")
+                          )
+                      ),
+                      new DoubleLeastPostAggregator(
+                          "a1",
+                          ImmutableList.of(
+                              new FieldAccessPostAggregator(null, "a1:0")
+                          )
+                      ),
+                      new DoubleLeastPostAggregator(
+                          "a2",
+                          ImmutableList.of(
+                              new FieldAccessPostAggregator(null, "a2:0")
+                          )
+                      )
+                                   )
+                  )
+                  .context(TIMESERIES_CONTEXT_DEFAULT)
+                  .build()
+        ),
+        ImmutableList.of(new Object[]{1L, 1.0F, 1.0D})
+    );
+  }
+
+  @Test
+  public void testGreatestCombinationPostAggregations() throws Exception
+  {
+    // Cannot vectorize due to virtual columns.
+    cannotVectorize();
+
+    testQuery(
+        "SELECT\n"
+        + "      greatest(cnt, 10, 10 * 2 + 3),\n"
+        + "      greatest(m1, 10.0, 10.2 * 2.0 + 3.0),\n"
+        + "      greatest(m2, 10.0, 10.2 * 2.0 + 3.0)\n"
+        + "      FROM \"foo\"\n",
+        ImmutableList.of(
+            Druids.newTimeseriesQueryBuilder()
+                  .dataSource(CalciteTests.DATASOURCE1)
+                  .intervals(querySegmentSpec(Filtration.eternity()))
+                  .granularity(Granularities.ALL)
+                  .aggregators(aggregators(
+                      new LongMaxAggregatorFactory("a0:0", "cnt"),
+                      new LongMaxAggregatorFactory("a0:1", null, "10", ExprMacroTable.nil()),
+                      new LongMaxAggregatorFactory("a0:2", null, "23", ExprMacroTable.nil()),
+                      new DoubleMaxAggregatorFactory("a1:0", "m1"),
+                      new DoubleMaxAggregatorFactory("a1:1", null, "10.0", ExprMacroTable.nil()),
+                      new DoubleMaxAggregatorFactory("a1:2", null, "23.4", ExprMacroTable.nil()),
+                      new DoubleMaxAggregatorFactory("a2:0", "m2"),
+                      new DoubleMaxAggregatorFactory("a2:1", null, "10.0", ExprMacroTable.nil()),
+                      new DoubleMaxAggregatorFactory("a2:2", null, "23.4", ExprMacroTable.nil())
+                  ))
+                  .postAggregators(ImmutableList.of(
+                      new LongGreatestPostAggregator(
+                          "a0",
+                          ImmutableList.of(
+                              new FieldAccessPostAggregator(null, "a0:0"),
+                              new FieldAccessPostAggregator(null, "a0:1"),
+                              new FieldAccessPostAggregator(null, "a0:2")
+                          )
+                      ),
+                      new DoubleGreatestPostAggregator(
+                          "a1",
+                          ImmutableList.of(
+                              new FieldAccessPostAggregator(null, "a1:0"),
+                              new FieldAccessPostAggregator(null, "a1:1"),
+                              new FieldAccessPostAggregator(null, "a1:2")
+                          )
+                      ),
+                      new DoubleGreatestPostAggregator(
+                          "a2",
+                          ImmutableList.of(
+                              new FieldAccessPostAggregator(null, "a2:0"),
+                              new FieldAccessPostAggregator(null, "a2:1"),
+                              new FieldAccessPostAggregator(null, "a2:2")
+                          )
+                      ))
+                  )
+                  .context(TIMESERIES_CONTEXT_DEFAULT)
+                  .build()
+        ),
+        ImmutableList.of(new Object[]{23L, 23.4D, 23.4D})
+    );
+  }
+
+  @Test
+  public void testLeastCombinationPostAggregations() throws Exception
+  {
+    // Cannot vectorize due to virtual columns.
+    cannotVectorize();
+
+    testQuery(
+        "SELECT\n"
+        + "      least(cnt, 10, 10 * 2 + 3),\n"
+        + "      least(m1, 10.0, 10.2 * 2.0 + 3.0),\n"
+        + "      least(m2, 10.0, 10.2 * 2.0 + 3.0)\n"
+        + "      FROM \"foo\"\n",
+        ImmutableList.of(
+            Druids.newTimeseriesQueryBuilder()
+                  .dataSource(CalciteTests.DATASOURCE1)
+                  .intervals(querySegmentSpec(Filtration.eternity()))
+                  .granularity(Granularities.ALL)
+                  .aggregators(aggregators(
+                      new LongMinAggregatorFactory("a0:0", "cnt"),
+                      new LongMinAggregatorFactory("a0:1", null, "10", ExprMacroTable.nil()),
+                      new LongMinAggregatorFactory("a0:2", null, "23", ExprMacroTable.nil()),
+                      new DoubleMinAggregatorFactory("a1:0", "m1"),
+                      new DoubleMinAggregatorFactory("a1:1", null, "10.0", ExprMacroTable.nil()),
+                      new DoubleMinAggregatorFactory("a1:2", null, "23.4", ExprMacroTable.nil()),
+                      new DoubleMinAggregatorFactory("a2:0", "m2"),
+                      new DoubleMinAggregatorFactory("a2:1", null, "10.0", ExprMacroTable.nil()),
+                      new DoubleMinAggregatorFactory("a2:2", null, "23.4", ExprMacroTable.nil())
+                  ))
+                  .postAggregators(ImmutableList.of(
+                      new LongLeastPostAggregator(
+                          "a0",
+                          ImmutableList.of(
+                              new FieldAccessPostAggregator(null, "a0:0"),
+                              new FieldAccessPostAggregator(null, "a0:1"),
+                              new FieldAccessPostAggregator(null, "a0:2")
+                          )
+                      ),
+                      new DoubleLeastPostAggregator(
+                          "a1",
+                          ImmutableList.of(
+                              new FieldAccessPostAggregator(null, "a1:0"),
+                              new FieldAccessPostAggregator(null, "a1:1"),
+                              new FieldAccessPostAggregator(null, "a1:2")
+                          )
+                      ),
+                      new DoubleLeastPostAggregator(
+                          "a2",
+                          ImmutableList.of(
+                              new FieldAccessPostAggregator(null, "a2:0"),
+                              new FieldAccessPostAggregator(null, "a2:1"),
+                              new FieldAccessPostAggregator(null, "a2:2")
+                          )
+                      ))
+                  )
+                  .context(TIMESERIES_CONTEXT_DEFAULT)
+                  .build()
+        ),
+        ImmutableList.of(new Object[]{1L, 1.0D, 1.0D})
+    );
+  }
+
+  @Test
+  public void testGreatestInvalidPostAggregations() throws Exception
+  {
+    // Cannot vectorize due to virtual columns.
+    cannotVectorize();
+    expectedException.expect(RuntimeException.class);
+    expectedException.expectCause(CoreMatchers.instanceOf(ISE.class));
+    expectedException.expectCause(
+        ThrowableMessageMatcher.hasMessage(
+            CoreMatchers.containsString(
+                "Cannot create aggregator factory for type[STRING]"
+            )
+        )
+    );
+
+    testQuery("SELECT GREATEST(dim1) FROM druid.foo", ImmutableList.of(), ImmutableList.of());
+  }
+
+  @Test
+  public void testLeastInvalidPostAggregations() throws Exception
+  {
+    // Cannot vectorize due to virtual columns.
+    cannotVectorize();
+    expectedException.expect(RuntimeException.class);
+    expectedException.expectCause(CoreMatchers.instanceOf(ISE.class));
+    expectedException.expectCause(
+        ThrowableMessageMatcher.hasMessage(
+            CoreMatchers.containsString(
+                "Cannot create aggregator factory for type[STRING]"
+            )
+        )
+    );
+
+    testQuery("SELECT LEAST(dim1) FROM druid.foo", ImmutableList.of(), ImmutableList.of());
+  }
+
+  @Test
+  public void testGreatestInvalidCombinationPostAggregations() throws Exception
+  {
+    // Cannot vectorize due to virtual columns.
+    cannotVectorize();
+    expectedException.expect(ValidationException.class);
+    expectedException.expectCause(CoreMatchers.instanceOf(IllegalArgumentException.class));
+    expectedException.expectCause(
+        ThrowableMessageMatcher.hasMessage(
+            CoreMatchers.containsString(
+                "Cannot infer return type for GREATEST; operand types: [INTEGER, VARCHAR]"
+            )
+        )
+    );
+
+    testQuery("SELECT GREATEST(10, dim1) FROM druid.foo", ImmutableList.of(), ImmutableList.of());
+  }
+
+  @Test
+  public void testLeastInvalidCombinationPostAggregations() throws Exception
+  {
+    // Cannot vectorize due to virtual columns.
+    cannotVectorize();
+    expectedException.expect(ValidationException.class);
+    expectedException.expectCause(CoreMatchers.instanceOf(IllegalArgumentException.class));
+    expectedException.expectCause(
+        ThrowableMessageMatcher.hasMessage(
+            CoreMatchers.containsString(
+                "Cannot infer return type for LEAST; operand types: [INTEGER, VARCHAR]"
+            )
+        )
+    );
+
+    testQuery("SELECT LEAST(10, dim1) FROM druid.foo", ImmutableList.of(), ImmutableList.of());
+  }
+
   @Test
   public void testAvgDailyCountDistinct() throws Exception
   {