Star Tree Min and Max Value Aggregators (#14625)

---------

Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>
(cherry picked from commit e21f6b8)
Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

server/src/main/java/org/opensearch/index/compositeindex/datacube/startree/aggregators/CountValueAggregator.java

@@ -7,70 +7,66 @@
  */
 package org.opensearch.index.compositeindex.datacube.startree.aggregators;
 
-import org.opensearch.index.compositeindex.datacube.MetricStat;
 import org.opensearch.index.compositeindex.datacube.startree.aggregators.numerictype.StarTreeNumericType;
 
 /**
  * Count value aggregator for star tree
  *
  * @opensearch.experimental
  */
-public class CountValueAggregator implements ValueAggregator<Long> {
-    public static final StarTreeNumericType VALUE_AGGREGATOR_TYPE = StarTreeNumericType.LONG;
+class CountValueAggregator implements ValueAggregator<Long> {
+
     public static final long DEFAULT_INITIAL_VALUE = 1L;
-    private StarTreeNumericType starTreeNumericType;
+    private final StarTreeNumericType starTreeNumericType;
+    private static final StarTreeNumericType VALUE_AGGREGATOR_TYPE = StarTreeNumericType.LONG;
 
     public CountValueAggregator(StarTreeNumericType starTreeNumericType) {
         this.starTreeNumericType = starTreeNumericType;
     }
 
-    @Override
-    public MetricStat getAggregationType() {
-        return MetricStat.COUNT;
-    }
-
     @Override
     public StarTreeNumericType getAggregatedValueType() {
         return VALUE_AGGREGATOR_TYPE;
     }
 
     @Override
     public Long getInitialAggregatedValueForSegmentDocValue(Long segmentDocValue) {
+
+        if (segmentDocValue == null) {
+            return getIdentityMetricValue();
+        }
+
         return DEFAULT_INITIAL_VALUE;
     }
 
     @Override
     public Long mergeAggregatedValueAndSegmentValue(Long value, Long segmentDocValue) {
-        return value + 1;
+        assert value != null;
+        if (segmentDocValue != null) {
+            return value + 1;
+        }
+        return value;
     }
 
     @Override
     public Long mergeAggregatedValues(Long value, Long aggregatedValue) {
+        if (value == null) {
+            value = getIdentityMetricValue();
+        }
+        if (aggregatedValue == null) {
+            aggregatedValue = getIdentityMetricValue();
+        }
         return value + aggregatedValue;
     }
 
-    @Override
-    public Long getInitialAggregatedValue(Long value) {
-        return value;
-    }
-
-    @Override
-    public int getMaxAggregatedValueByteSize() {
-        return Long.BYTES;
-    }
-
-    @Override
-    public Long toLongValue(Long value) {
-        return value;
-    }
-
     @Override
     public Long toStarTreeNumericTypeValue(Long value) {
         return value;
     }
 
     @Override
     public Long getIdentityMetricValue() {
+        // in present aggregations, if the metric behind count is missing, we treat it as 0
         return 0L;
     }
 }

server/src/main/java/org/opensearch/index/compositeindex/datacube/startree/aggregators/MaxValueAggregator.java

@@ -0,0 +1,27 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+package org.opensearch.index.compositeindex.datacube.startree.aggregators;
+
+import org.opensearch.index.compositeindex.datacube.startree.aggregators.numerictype.StarTreeNumericType;
+
+/**
+ * Max value aggregator for star tree
+ *
+ * @opensearch.experimental
+ */
+class MaxValueAggregator extends StatelessDoubleValueAggregator {
+
+    public MaxValueAggregator(StarTreeNumericType starTreeNumericType) {
+        super(starTreeNumericType, null);
+    }
+
+    @Override
+    protected Double performValueAggregation(Double aggregatedValue, Double segmentDocValue) {
+        return Math.max(aggregatedValue, segmentDocValue);
+    }
+}

server/src/main/java/org/opensearch/index/compositeindex/datacube/startree/aggregators/MetricAggregatorInfo.java

@@ -16,6 +16,7 @@
 
 /**
  * Builds aggregation function and doc values field pair to support various aggregations
+ *
  * @opensearch.experimental
  */
 public class MetricAggregatorInfo implements Comparable<MetricAggregatorInfo> {
@@ -79,7 +80,15 @@ public StarTreeNumericType getAggregatedValueType() {
      * @return field name with metric type and field
      */
     public String toFieldName() {
-        return starFieldName + DELIMITER + field + DELIMITER + metricStat.getTypeName();
+        return toFieldName(starFieldName, field, metricStat.getTypeName());
+
+    }
+
+    /**
+     * @return field name with star-tree field name metric type and field
+     */
+    public static String toFieldName(String starFieldName, String field, String metricName) {
+        return starFieldName + DELIMITER + field + DELIMITER + metricName;
     }
 
     @Override
@@ -94,7 +103,7 @@ public boolean equals(Object obj) {
         }
         if (obj instanceof MetricAggregatorInfo) {
             MetricAggregatorInfo anotherPair = (MetricAggregatorInfo) obj;
-            return metricStat == anotherPair.metricStat && field.equals(anotherPair.field);
+            return metricStat.equals(anotherPair.metricStat) && field.equals(anotherPair.field);
         }
         return false;
     }

server/src/main/java/org/opensearch/index/compositeindex/datacube/startree/aggregators/MinValueAggregator.java

@@ -0,0 +1,27 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+package org.opensearch.index.compositeindex.datacube.startree.aggregators;
+
+import org.opensearch.index.compositeindex.datacube.startree.aggregators.numerictype.StarTreeNumericType;
+
+/**
+ * Min value aggregator for star tree
+ *
+ * @opensearch.experimental
+ */
+class MinValueAggregator extends StatelessDoubleValueAggregator {
+
+    public MinValueAggregator(StarTreeNumericType starTreeNumericType) {
+        super(starTreeNumericType, null);
+    }
+
+    @Override
+    protected Double performValueAggregation(Double aggregatedValue, Double segmentDocValue) {
+        return Math.min(aggregatedValue, segmentDocValue);
+    }
+}

server/src/main/java/org/opensearch/index/compositeindex/datacube/startree/aggregators/StatelessDoubleValueAggregator.java

@@ -0,0 +1,81 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+package org.opensearch.index.compositeindex.datacube.startree.aggregators;
+
+import org.opensearch.index.compositeindex.datacube.startree.aggregators.numerictype.StarTreeNumericType;
+
+/**
+ * This is an abstract class that defines the common methods for all double value aggregators
+ * It is stateless.
+ *
+ * @opensearch.experimental
+ */
+abstract class StatelessDoubleValueAggregator implements ValueAggregator<Double> {
+
+    protected final StarTreeNumericType starTreeNumericType;
+    protected final Double identityValue;
+    private static final StarTreeNumericType VALUE_AGGREGATOR_TYPE = StarTreeNumericType.DOUBLE;
+
+    public StatelessDoubleValueAggregator(StarTreeNumericType starTreeNumericType, Double identityValue) {
+        this.starTreeNumericType = starTreeNumericType;
+        this.identityValue = identityValue;
+    }
+
+    @Override
+    public StarTreeNumericType getAggregatedValueType() {
+        return VALUE_AGGREGATOR_TYPE;
+    }
+
+    @Override
+    public Double getInitialAggregatedValueForSegmentDocValue(Long segmentDocValue) {
+        if (segmentDocValue == null) {
+            return getIdentityMetricValue();
+        }
+        return starTreeNumericType.getDoubleValue(segmentDocValue);
+    }
+
+    @Override
+    public Double mergeAggregatedValues(Double value, Double aggregatedValue) {
+        if (value == null && aggregatedValue != null) {
+            return aggregatedValue;
+        } else if (value != null && aggregatedValue == null) {
+            return value;
+        } else if (value == null) {
+            return getIdentityMetricValue();
+        }
+        return performValueAggregation(value, aggregatedValue);
+    }
+
+    @Override
+    public Double toStarTreeNumericTypeValue(Long value) {
+        try {
+            if (value == null) {
+                return getIdentityMetricValue();
+            }
+            return starTreeNumericType.getDoubleValue(value);
+        } catch (Exception e) {
+            throw new IllegalStateException("Cannot convert " + value + " to sortable aggregation type", e);
+        }
+    }
+
+    @Override
+    public Double getIdentityMetricValue() {
+        // the identity value that we return should be inline with the existing aggregations
+        return identityValue;
+    }
+
+    /**
+     * Performs stateless aggregation on the value and the segmentDocValue based on the implementation
+     *
+     * @param aggregatedValue aggregated value for the segment so far
+     * @param segmentDocValue current segment doc value
+     * @return aggregated value
+     */
+    protected abstract Double performValueAggregation(Double aggregatedValue, Double segmentDocValue);
+
+}