Support aggregations min, max (#541)

* Support aggregations min, max * update * support string in max and min * update * update * added min/max support for date, datetime, time, timestamp types * address comments * update * update * address comment * add comparison tests * update * added doc
opendistro-for-elasticsearch · Sep 30, 2020 · a0e9b5c · a0e9b5c
1 parent 87e7e02
commit a0e9b5c
Show file tree

Hide file tree

Showing 23 changed files with 710 additions and 36 deletions.
diff --git a/...src/main/java/com/amazon/opendistroforelasticsearch/sql/data/model/AbstractExprValue.java b/...src/main/java/com/amazon/opendistroforelasticsearch/sql/data/model/AbstractExprValue.java
@@ -30,7 +30,7 @@ public abstract class AbstractExprValue implements ExprValue {
   public int compareTo(ExprValue other) {
     if (this.isNull() || this.isMissing() || other.isNull() || other.isMissing()) {
       throw new IllegalStateException(
-          String.format("[BUG] Unreachable, Comparing with NULL or MISSING is undefined"));
+           String.format("[BUG] Unreachable, Comparing with NULL or MISSING is undefined"));
     }
     if ((this.isNumber() && other.isNumber()) || this.type() == other.type()) {
       return compare(other);

diff --git a/...rc/main/java/com/amazon/opendistroforelasticsearch/sql/data/model/ExprTimestampValue.java b/...rc/main/java/com/amazon/opendistroforelasticsearch/sql/data/model/ExprTimestampValue.java
@@ -101,12 +101,12 @@ public String toString() {
 
   @Override
   public int compare(ExprValue other) {
-    return timestamp.compareTo(other.timestampValue());
+    return timestamp.compareTo(other.timestampValue().atZone(ZONE).toInstant());
   }
 
   @Override
   public boolean equal(ExprValue other) {
-    return timestamp.equals(other.timestampValue());
+    return timestamp.equals(other.timestampValue().atZone(ZONE).toInstant());
   }
 
   @Override

diff --git a/core/src/main/java/com/amazon/opendistroforelasticsearch/sql/expression/DSL.java b/core/src/main/java/com/amazon/opendistroforelasticsearch/sql/expression/DSL.java
@@ -453,6 +453,14 @@ public Aggregator count(Expression... expressions) {
     return aggregate(BuiltinFunctionName.COUNT, expressions);
   }
 
+  public Aggregator min(Expression... expressions) {
+    return aggregate(BuiltinFunctionName.MIN, expressions);
+  }
+
+  public Aggregator max(Expression... expressions) {
+    return aggregate(BuiltinFunctionName.MAX, expressions);
+  }
+
   private FunctionExpression function(BuiltinFunctionName functionName, Expression... expressions) {
     return (FunctionExpression) repository.compile(
         functionName.getName(), Arrays.asList(expressions));

diff --git a/.../com/amazon/opendistroforelasticsearch/sql/expression/aggregation/AggregatorFunction.java b/.../com/amazon/opendistroforelasticsearch/sql/expression/aggregation/AggregatorFunction.java
@@ -17,12 +17,16 @@
 
 import static com.amazon.opendistroforelasticsearch.sql.data.type.ExprCoreType.ARRAY;
 import static com.amazon.opendistroforelasticsearch.sql.data.type.ExprCoreType.BOOLEAN;
+import static com.amazon.opendistroforelasticsearch.sql.data.type.ExprCoreType.DATE;
+import static com.amazon.opendistroforelasticsearch.sql.data.type.ExprCoreType.DATETIME;
 import static com.amazon.opendistroforelasticsearch.sql.data.type.ExprCoreType.DOUBLE;
 import static com.amazon.opendistroforelasticsearch.sql.data.type.ExprCoreType.FLOAT;
 import static com.amazon.opendistroforelasticsearch.sql.data.type.ExprCoreType.INTEGER;
 import static com.amazon.opendistroforelasticsearch.sql.data.type.ExprCoreType.LONG;
 import static com.amazon.opendistroforelasticsearch.sql.data.type.ExprCoreType.STRING;
 import static com.amazon.opendistroforelasticsearch.sql.data.type.ExprCoreType.STRUCT;
+import static com.amazon.opendistroforelasticsearch.sql.data.type.ExprCoreType.TIME;
+import static com.amazon.opendistroforelasticsearch.sql.data.type.ExprCoreType.TIMESTAMP;
 
 import com.amazon.opendistroforelasticsearch.sql.expression.function.BuiltinFunctionName;
 import com.amazon.opendistroforelasticsearch.sql.expression.function.BuiltinFunctionRepository;
@@ -52,6 +56,8 @@ public static void register(BuiltinFunctionRepository repository) {
     repository.register(avg());
     repository.register(sum());
     repository.register(count());
+    repository.register(min());
+    repository.register(max());
   }
 
   private static FunctionResolver avg() {
@@ -106,4 +112,57 @@ private static FunctionResolver sum() {
             .build()
     );
   }
+
+  private static FunctionResolver min() {
+    FunctionName functionName = BuiltinFunctionName.MIN.getName();
+    return new FunctionResolver(
+        functionName,
+        new ImmutableMap.Builder<FunctionSignature, FunctionBuilder>()
+            .put(new FunctionSignature(functionName, Collections.singletonList(INTEGER)),
+                arguments -> new MinAggregator(arguments, INTEGER))
+            .put(new FunctionSignature(functionName, Collections.singletonList(LONG)),
+                arguments -> new MinAggregator(arguments, LONG))
+            .put(new FunctionSignature(functionName, Collections.singletonList(FLOAT)),
+                arguments -> new MinAggregator(arguments, FLOAT))
+            .put(new FunctionSignature(functionName, Collections.singletonList(DOUBLE)),
+                arguments -> new MinAggregator(arguments, DOUBLE))
+            .put(new FunctionSignature(functionName, Collections.singletonList(STRING)),
+                arguments -> new MinAggregator(arguments, STRING))
+            .put(new FunctionSignature(functionName, Collections.singletonList(DATE)),
+                arguments -> new MinAggregator(arguments, DATE))
+            .put(new FunctionSignature(functionName, Collections.singletonList(DATETIME)),
+                arguments -> new MinAggregator(arguments, DATETIME))
+            .put(new FunctionSignature(functionName, Collections.singletonList(TIME)),
+                arguments -> new MinAggregator(arguments, TIME))
+            .put(new FunctionSignature(functionName, Collections.singletonList(TIMESTAMP)),
+                arguments -> new MinAggregator(arguments, TIMESTAMP))
+            .build());
+  }
+
+  private static FunctionResolver max() {
+    FunctionName functionName = BuiltinFunctionName.MAX.getName();
+    return new FunctionResolver(
+        functionName,
+        new ImmutableMap.Builder<FunctionSignature, FunctionBuilder>()
+            .put(new FunctionSignature(functionName, Collections.singletonList(INTEGER)),
+                arguments -> new MaxAggregator(arguments, INTEGER))
+            .put(new FunctionSignature(functionName, Collections.singletonList(LONG)),
+                arguments -> new MaxAggregator(arguments, LONG))
+            .put(new FunctionSignature(functionName, Collections.singletonList(FLOAT)),
+                arguments -> new MaxAggregator(arguments, FLOAT))
+            .put(new FunctionSignature(functionName, Collections.singletonList(DOUBLE)),
+                arguments -> new MaxAggregator(arguments, DOUBLE))
+            .put(new FunctionSignature(functionName, Collections.singletonList(STRING)),
+                arguments -> new MaxAggregator(arguments, STRING))
+            .put(new FunctionSignature(functionName, Collections.singletonList(DATE)),
+                arguments -> new MaxAggregator(arguments, DATE))
+            .put(new FunctionSignature(functionName, Collections.singletonList(DATETIME)),
+                arguments -> new MaxAggregator(arguments, DATETIME))
+            .put(new FunctionSignature(functionName, Collections.singletonList(TIME)),
+                arguments -> new MaxAggregator(arguments, TIME))
+            .put(new FunctionSignature(functionName, Collections.singletonList(TIMESTAMP)),
+                arguments -> new MaxAggregator(arguments, TIMESTAMP))
+            .build()
+    );
+  }
 }
diff --git a/.../java/com/amazon/opendistroforelasticsearch/sql/expression/aggregation/AvgAggregator.java b/.../java/com/amazon/opendistroforelasticsearch/sql/expression/aggregation/AvgAggregator.java
@@ -46,9 +46,7 @@ public AvgState create() {
   public AvgState iterate(BindingTuple tuple, AvgState state) {
     Expression expression = getArguments().get(0);
     ExprValue value = expression.valueOf(tuple);
-    if (value.isNull() || value.isMissing()) {
-      state.isNullResult = true;
-    } else {
+    if (!(value.isNull() || value.isMissing())) {
       state.count++;
       state.total += ExprValueUtils.getDoubleValue(value);
     }
@@ -63,19 +61,18 @@ public String toString() {
   /**
    * Average State.
    */
-  protected class AvgState implements AggregationState {
+  protected static class AvgState implements AggregationState {
     private int count;
     private double total;
-    private boolean isNullResult = false;
 
-    public AvgState() {
+    AvgState() {
       this.count = 0;
       this.total = 0d;
     }
 
     @Override
     public ExprValue result() {
-      return isNullResult ? ExprNullValue.of() : ExprValueUtils.doubleValue(total / count);
+      return count == 0 ? ExprNullValue.of() : ExprValueUtils.doubleValue(total / count);
     }
   }
 }
diff --git a/...ava/com/amazon/opendistroforelasticsearch/sql/expression/aggregation/CountAggregator.java b/...ava/com/amazon/opendistroforelasticsearch/sql/expression/aggregation/CountAggregator.java
@@ -56,10 +56,10 @@ public String toString() {
   /**
    * Count State.
    */
-  protected class CountState implements AggregationState {
+  protected static class CountState implements AggregationState {
     private int count;
 
-    public CountState() {
+    CountState() {
       this.count = 0;
     }
 

diff --git a/.../java/com/amazon/opendistroforelasticsearch/sql/expression/aggregation/MaxAggregator.java b/.../java/com/amazon/opendistroforelasticsearch/sql/expression/aggregation/MaxAggregator.java
@@ -0,0 +1,72 @@
+/*
+ *   Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License").
+ *   You may not use this file except in compliance with the License.
+ *   A copy of the License is located at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   or in the "license" file accompanying this file. This file 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 com.amazon.opendistroforelasticsearch.sql.expression.aggregation;
+
+import static com.amazon.opendistroforelasticsearch.sql.data.model.ExprValueUtils.LITERAL_NULL;
+import static com.amazon.opendistroforelasticsearch.sql.utils.ExpressionUtils.format;
+
+import com.amazon.opendistroforelasticsearch.sql.data.model.ExprValue;
+import com.amazon.opendistroforelasticsearch.sql.data.type.ExprCoreType;
+import com.amazon.opendistroforelasticsearch.sql.expression.Expression;
+import com.amazon.opendistroforelasticsearch.sql.expression.function.BuiltinFunctionName;
+import com.amazon.opendistroforelasticsearch.sql.storage.bindingtuple.BindingTuple;
+import java.util.List;
+
+public class MaxAggregator extends Aggregator<MaxAggregator.MaxState> {
+
+  public MaxAggregator(List<Expression> arguments, ExprCoreType returnType) {
+    super(BuiltinFunctionName.MAX.getName(), arguments, returnType);
+  }
+
+  @Override
+  public MaxState create() {
+    return new MaxState();
+  }
+
+  @Override
+  public MaxState iterate(BindingTuple tuple, MaxState state) {
+    Expression expression = getArguments().get(0);
+    ExprValue value = expression.valueOf(tuple);
+    if (!(value.isNull() || value.isMissing())) {
+      state.max(value);
+    }
+    return state;
+  }
+
+  @Override
+  public String toString() {
+    return String.format("max(%s)", format(getArguments()));
+  }
+
+  protected static class MaxState implements AggregationState {
+    private ExprValue maxResult;
+
+    MaxState() {
+      maxResult = LITERAL_NULL;
+    }
+
+    public void max(ExprValue value) {
+      maxResult = maxResult.isNull() ? value
+          : (maxResult.compareTo(value) > 0)
+          ? maxResult : value;
+    }
+
+    @Override
+    public ExprValue result() {
+      return maxResult;
+    }
+  }
+}
diff --git a/.../java/com/amazon/opendistroforelasticsearch/sql/expression/aggregation/MinAggregator.java b/.../java/com/amazon/opendistroforelasticsearch/sql/expression/aggregation/MinAggregator.java
@@ -0,0 +1,77 @@
+/*
+ *   Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License").
+ *   You may not use this file except in compliance with the License.
+ *   A copy of the License is located at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   or in the "license" file accompanying this file. This file 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 com.amazon.opendistroforelasticsearch.sql.expression.aggregation;
+
+import static com.amazon.opendistroforelasticsearch.sql.data.model.ExprValueUtils.LITERAL_NULL;
+import static com.amazon.opendistroforelasticsearch.sql.utils.ExpressionUtils.format;
+
+import com.amazon.opendistroforelasticsearch.sql.data.model.ExprValue;
+import com.amazon.opendistroforelasticsearch.sql.data.type.ExprCoreType;
+import com.amazon.opendistroforelasticsearch.sql.expression.Expression;
+import com.amazon.opendistroforelasticsearch.sql.expression.function.BuiltinFunctionName;
+import com.amazon.opendistroforelasticsearch.sql.storage.bindingtuple.BindingTuple;
+import java.util.List;
+
+/**
+ * The minimum aggregator aggregate the value evaluated by the expression.
+ * If the expression evaluated result is NULL or MISSING, then the result is NULL.
+ */
+public class MinAggregator extends Aggregator<MinAggregator.MinState> {
+
+  public MinAggregator(List<Expression> arguments, ExprCoreType returnType) {
+    super(BuiltinFunctionName.MIN.getName(), arguments, returnType);
+  }
+
+
+  @Override
+  public MinState create() {
+    return new MinState();
+  }
+
+  @Override
+  public MinState iterate(BindingTuple tuple, MinState state) {
+    Expression expression = getArguments().get(0);
+    ExprValue value = expression.valueOf(tuple);
+    if (!(value.isNull() || value.isMissing())) {
+      state.min(value);
+    }
+    return state;
+  }
+
+  @Override
+  public String toString() {
+    return String.format("min(%s)", format(getArguments()));
+  }
+
+  protected static class MinState implements AggregationState {
+    private ExprValue minResult;
+
+    MinState() {
+      minResult = LITERAL_NULL;
+    }
+
+    public void min(ExprValue value) {
+      minResult = minResult.isNull() ? value
+          : (minResult.compareTo(value) < 0)
+          ? minResult : value;
+    }
+
+    @Override
+    public ExprValue result() {
+      return minResult;
+    }
+  }
+}
diff --git a/.../java/com/amazon/opendistroforelasticsearch/sql/expression/aggregation/SumAggregator.java b/.../java/com/amazon/opendistroforelasticsearch/sql/expression/aggregation/SumAggregator.java
@@ -56,9 +56,8 @@ public SumState create() {
   public SumState iterate(BindingTuple tuple, SumState state) {
     Expression expression = getArguments().get(0);
     ExprValue value = expression.valueOf(tuple);
-    if (value.isNull() || value.isMissing()) {
-      state.isNullResult = true;
-    } else {
+    if (!(value.isNull() || value.isMissing())) {
+      state.isEmptyCollection = false;
       state.add(value);
     }
     return state;
@@ -72,15 +71,16 @@ public String toString() {
   /**
    * Sum State.
    */
-  protected class SumState implements AggregationState {
+  protected static class SumState implements AggregationState {
 
     private final ExprCoreType type;
     private ExprValue sumResult;
-    private boolean isNullResult = false;
+    private boolean isEmptyCollection;
 
-    public SumState(ExprCoreType type) {
+    SumState(ExprCoreType type) {
       this.type = type;
       sumResult = ExprValueUtils.integerValue(0);
+      isEmptyCollection = true;
     }
 
     /**
@@ -108,7 +108,7 @@ public void add(ExprValue value) {
 
     @Override
     public ExprValue result() {
-      return isNullResult ? ExprNullValue.of() : sumResult;
+      return isEmptyCollection ? ExprNullValue.of() : sumResult;
     }
   }
 }
diff --git a/...va/com/amazon/opendistroforelasticsearch/sql/expression/function/BuiltinFunctionName.java b/...va/com/amazon/opendistroforelasticsearch/sql/expression/function/BuiltinFunctionName.java
@@ -111,6 +111,8 @@ public enum BuiltinFunctionName {
   AVG(FunctionName.of("avg")),
   SUM(FunctionName.of("sum")),
   COUNT(FunctionName.of("count")),
+  MIN(FunctionName.of("min")),
+  MAX(FunctionName.of("max")),
 
   /**
    * Text Functions.