Fix incorrect results returned by min, max and avg

integ-test/src/test/java/org/opensearch/sql/legacy/SQLIntegTestCase.java

@@ -48,6 +48,7 @@
 import static org.opensearch.sql.legacy.TestUtils.getGameOfThronesIndexMapping;
 import static org.opensearch.sql.legacy.TestUtils.getJoinTypeIndexMapping;
 import static org.opensearch.sql.legacy.TestUtils.getLocationIndexMapping;
+import static org.opensearch.sql.legacy.TestUtils.getMappingFile;
 import static org.opensearch.sql.legacy.TestUtils.getNestedSimpleIndexMapping;
 import static org.opensearch.sql.legacy.TestUtils.getNestedTypeIndexMapping;
 import static org.opensearch.sql.legacy.TestUtils.getOdbcIndexMapping;
@@ -575,7 +576,11 @@ public enum Index {
     BEER(TestsConstants.TEST_INDEX_BEER,
         "beer",
         null,
-        "src/test/resources/beer.stackexchange.json"),;
+        "src/test/resources/beer.stackexchange.json"),
+    NULL_MISSING(TestsConstants.TEST_INDEX_NULL_MISSING,
+        "null_missing",
+        getMappingFile("null_missing_index_mapping.json"),
+        "src/test/resources/null_missing.json"),;
 
     private final String name;
     private final String type;

integ-test/src/test/java/org/opensearch/sql/legacy/TestsConstants.java

@@ -51,6 +51,7 @@ public class TestsConstants {
   public final static String TEST_INDEX_DATATYPE_NUMERIC = TEST_INDEX + "_datatypes_numeric";
   public final static String TEST_INDEX_DATATYPE_NONNUMERIC = TEST_INDEX + "_datatypes_nonnumeric";
   public final static String TEST_INDEX_BEER = TEST_INDEX + "_beer";
+  public final static String TEST_INDEX_NULL_MISSING = TEST_INDEX + "_null_missing";
 
   public final static String DATE_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'";
   public final static String TS_DATE_FORMAT = "yyyy-MM-dd HH:mm:ss.SSS";

integ-test/src/test/java/org/opensearch/sql/sql/AggregationIT.java

@@ -6,36 +6,127 @@
 package org.opensearch.sql.sql;
 
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_BANK;
+import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_NULL_MISSING;
+import static org.opensearch.sql.legacy.plugin.RestSqlAction.QUERY_API_ENDPOINT;
 import static org.opensearch.sql.util.MatcherUtils.rows;
 import static org.opensearch.sql.util.MatcherUtils.schema;
 import static org.opensearch.sql.util.MatcherUtils.verifyDataRows;
 import static org.opensearch.sql.util.MatcherUtils.verifySchema;
+import static org.opensearch.sql.util.TestUtils.getResponseBody;
 
 import java.io.IOException;
+import java.util.Locale;
 import org.json.JSONObject;
 import org.junit.jupiter.api.Test;
+import org.opensearch.client.Request;
+import org.opensearch.client.RequestOptions;
+import org.opensearch.client.Response;
 import org.opensearch.sql.legacy.SQLIntegTestCase;
 
 public class AggregationIT extends SQLIntegTestCase {
   @Override
   protected void init() throws Exception {
+    super.init();
     loadIndex(Index.BANK);
+    loadIndex(Index.NULL_MISSING);
   }
 
   @Test
-  void filteredAggregatePushedDown() throws IOException {
+  public void testFilteredAggregatePushedDown() throws IOException {
     JSONObject response = executeQuery(
         "SELECT COUNT(*) FILTER(WHERE age > 35) FROM " + TEST_INDEX_BANK);
-    verifySchema(response, schema("COUNT(*)", null, "integer"));
+    verifySchema(response, schema("COUNT(*) FILTER(WHERE age > 35)", null, "integer"));
     verifyDataRows(response, rows(3));
   }
 
   @Test
-  void filteredAggregateNotPushedDown() throws IOException {
+  public void testFilteredAggregateNotPushedDown() throws IOException {
     JSONObject response = executeQuery(
         "SELECT COUNT(*) FILTER(WHERE age > 35) FROM (SELECT * FROM " + TEST_INDEX_BANK
             + ") AS a");
-    verifySchema(response, schema("COUNT(*)", null, "integer"));
+    verifySchema(response, schema("COUNT(*) FILTER(WHERE age > 35)", null, "integer"));
     verifyDataRows(response, rows(3));
   }
+
+  @Test
+  public void testPushedDownAggregationOnNullValues() throws IOException {
+    // OpenSearch aggregation query (MetricAggregation)
+    var response = executeQuery(String.format(
+        "SELECT min(`int`), max(`int`), avg(`int`), min(`dbl`), max(`dbl`), avg(`dbl`) " +
+        "FROM %s WHERE `key` = 'null'", TEST_INDEX_NULL_MISSING));
+    verifySchema(response,
+        schema("min(`int`)", null, "integer"), schema("max(`int`)", null, "integer"),
+        schema("avg(`int`)", null, "double"), schema("min(`dbl`)", null, "double"),
+        schema("max(`dbl`)", null, "double"), schema("avg(`dbl`)", null, "double"));
+    verifyDataRows(response, rows(null, null, null, null, null, null));
+  }
+
+  @Test
+  public void testPushedDownAggregationOnMissingValues() throws IOException {
+    // OpenSearch aggregation query (MetricAggregation)
+    var response = executeQuery(String.format(
+        "SELECT min(`int`), max(`int`), avg(`int`), min(`dbl`), max(`dbl`), avg(`dbl`) " +
+        "FROM %s WHERE `key` = 'null'", TEST_INDEX_NULL_MISSING));
+    verifySchema(response,
+        schema("min(`int`)", null, "integer"), schema("max(`int`)", null, "integer"),
+        schema("avg(`int`)", null, "double"), schema("min(`dbl`)", null, "double"),
+        schema("max(`dbl`)", null, "double"), schema("avg(`dbl`)", null, "double"));
+    verifyDataRows(response, rows(null, null, null, null, null, null));
+  }
+
+  @Test
+  public void testInMemoryAggregationOnNullValues() throws IOException {
+    // In-memory aggregation performed by the plugin
+    var response = executeQuery(String.format("SELECT"
+        + " min(`int`) over (PARTITION BY `key`), max(`int`) over (PARTITION BY `key`),"
+        + " avg(`int`) over (PARTITION BY `key`), min(`dbl`) over (PARTITION BY `key`),"
+        + " max(`dbl`) over (PARTITION BY `key`), avg(`dbl`) over (PARTITION BY `key`)"
+        + " FROM %s WHERE `key` = 'null'", TEST_INDEX_NULL_MISSING));
+    verifySchema(response,
+        schema("min(`int`) over (PARTITION BY `key`)", null, "integer"),
+        schema("max(`int`) over (PARTITION BY `key`)", null, "integer"),
+        schema("avg(`int`) over (PARTITION BY `key`)", null, "double"),
+        schema("min(`dbl`) over (PARTITION BY `key`)", null, "double"),
+        schema("max(`dbl`) over (PARTITION BY `key`)", null, "double"),
+        schema("avg(`dbl`) over (PARTITION BY `key`)", null, "double"));
+    verifyDataRows(response, // 4 rows with null values
+        rows(null, null, null, null, null, null),
+        rows(null, null, null, null, null, null),
+        rows(null, null, null, null, null, null),
+        rows(null, null, null, null, null, null));
+  }
+
+  @Test
+  public void testInMemoryAggregationOnMissingValues() throws IOException {
+    // In-memory aggregation performed by the plugin
+    var response = executeQuery(String.format("SELECT"
+        + " min(`int`) over (PARTITION BY `key`), max(`int`) over (PARTITION BY `key`),"
+        + " avg(`int`) over (PARTITION BY `key`), min(`dbl`) over (PARTITION BY `key`),"
+        + " max(`dbl`) over (PARTITION BY `key`), avg(`dbl`) over (PARTITION BY `key`)"
+        + " FROM %s WHERE `key` = 'missing'", TEST_INDEX_NULL_MISSING));
+    verifySchema(response,
+        schema("min(`int`) over (PARTITION BY `key`)", null, "integer"),
+        schema("max(`int`) over (PARTITION BY `key`)", null, "integer"),
+        schema("avg(`int`) over (PARTITION BY `key`)", null, "double"),
+        schema("min(`dbl`) over (PARTITION BY `key`)", null, "double"),
+        schema("max(`dbl`) over (PARTITION BY `key`)", null, "double"),
+        schema("avg(`dbl`) over (PARTITION BY `key`)", null, "double"));
+    verifyDataRows(response, // 4 rows with null values
+        rows(null, null, null, null, null, null),
+        rows(null, null, null, null, null, null),
+        rows(null, null, null, null, null, null),
+        rows(null, null, null, null, null, null));
+  }
+
+  protected JSONObject executeQuery(String query) throws IOException {
+    Request request = new Request("POST", QUERY_API_ENDPOINT);
+    request.setJsonEntity(String.format(Locale.ROOT, "{\n" + "  \"query\": \"%s\"\n" + "}", query));
+
+    RequestOptions.Builder restOptionsBuilder = RequestOptions.DEFAULT.toBuilder();
+    restOptionsBuilder.addHeader("Content-Type", "application/json");
+    request.setOptions(restOptionsBuilder);
+
+    Response response = client().performRequest(request);
+    return new JSONObject(getResponseBody(response));
+  }
 }

integ-test/src/test/resources/indexDefinitions/null_missing_index_mapping.json

@@ -0,0 +1,37 @@
+{
+    "mappings" : {
+        "properties" : {
+            "key" : {
+                "type" : "keyword"
+            },
+            "int" : {
+                "type" : "integer"
+            },
+            "dbl" : {
+                "type" : "double"
+            },
+            "bool" : {
+                "type" : "boolean"
+            },
+            "str" : {
+                "type" : "text"
+            },
+            "date" : {
+                "type" : "date",
+                "format": "yyyy-MM-dd"
+            },
+            "time" : {
+                "type" : "date",
+                "format": "HH:mm:ss"
+            },
+            "datetime" : {
+                "type" : "date",
+                "format": "yyyy-MM-dd HH:mm:ss"
+            },
+            "timestamp" : {
+                "type" : "date",
+                "format": "yyyy-MM-dd HH:mm:ss"
+            }
+        }
+    }
+}

integ-test/src/test/resources/null_missing.json

@@ -0,0 +1,18 @@
+{"index":{}}
+{"key" : "values", "int" : 42, "dbl" : 3.1415, "bool" : true, "str" : "pewpew", "date" : "1984-05-22", "time" : "22:13:37", "datetime" : "1984-05-22 22:13:37", "timestamp" : "2000-01-02 03:04:05" }
+{"index":{}}
+{"key" : "missing"}
+{"index":{}}
+{"key" : "missing"}
+{"index":{}}
+{"key" : "missing"}
+{"index":{}}
+{"key" : "missing"}
+{"index":{}}
+{"key" : "null", "int" : null, "dbl" : null, "bool" : null, "str" : null, "date" : null, "time" : null, "datetime" : null, "timestamp" : null}
+{"index":{}}
+{"key" : "null", "int" : null, "dbl" : null, "bool" : null, "str" : null, "date" : null, "time" : null, "datetime" : null, "timestamp" : null}
+{"index":{}}
+{"key" : "null", "int" : null, "dbl" : null, "bool" : null, "str" : null, "date" : null, "time" : null, "datetime" : null, "timestamp" : null}
+{"index":{}}
+{"key" : "null", "int" : null, "dbl" : null, "bool" : null, "str" : null, "date" : null, "time" : null, "datetime" : null, "timestamp" : null}

opensearch/src/main/java/org/opensearch/sql/opensearch/response/agg/SingleValueParser.java

@@ -13,7 +13,7 @@
 
 package org.opensearch.sql.opensearch.response.agg;
 
-import static org.opensearch.sql.opensearch.response.agg.Utils.handleNanValue;
+import static org.opensearch.sql.opensearch.response.agg.Utils.handleNanInfValue;
 
 import java.util.Collections;
 import java.util.Map;
@@ -34,6 +34,6 @@ public class SingleValueParser implements MetricParser {
   public Map<String, Object> parse(Aggregation agg) {
     return Collections.singletonMap(
         agg.getName(),
-        handleNanValue(((NumericMetricsAggregation.SingleValue) agg).value()));
+        handleNanInfValue(((NumericMetricsAggregation.SingleValue) agg).value()));
   }
 }

opensearch/src/main/java/org/opensearch/sql/opensearch/response/agg/StatsParser.java

@@ -13,7 +13,7 @@
 
 package org.opensearch.sql.opensearch.response.agg;
 
-import static org.opensearch.sql.opensearch.response.agg.Utils.handleNanValue;
+import static org.opensearch.sql.opensearch.response.agg.Utils.handleNanInfValue;
 
 import java.util.Collections;
 import java.util.Map;
@@ -36,6 +36,6 @@ public class StatsParser implements MetricParser {
   @Override
   public Map<String, Object> parse(Aggregation agg) {
     return Collections.singletonMap(
-        agg.getName(), handleNanValue(valueExtractor.apply((ExtendedStats) agg)));
+        agg.getName(), handleNanInfValue(valueExtractor.apply((ExtendedStats) agg)));
   }
 }

opensearch/src/main/java/org/opensearch/sql/opensearch/response/agg/Utils.java

@@ -18,10 +18,10 @@
 @UtilityClass
 public class Utils {
   /**
-   * Utils to handle Nan Value.
-   * @return null if is Nan.
+   * Utils to handle Nan/Infinite Value.
+   * @return null if is Nan or is +-Infinity.
    */
-  public static Object handleNanValue(double value) {
-    return Double.isNaN(value) ? null : value;
+  public static Object handleNanInfValue(double value) {
+    return Double.isNaN(value) || Double.isInfinite(value) ? null : value;
   }
 }

opensearch/src/test/java/org/opensearch/sql/opensearch/response/OpenSearchAggregationResponseParserTest.java

@@ -13,7 +13,7 @@
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.opensearch.sql.opensearch.response.AggregationResponseUtils.fromJson;
-import static org.opensearch.sql.opensearch.response.agg.Utils.handleNanValue;
+import static org.opensearch.sql.opensearch.response.agg.Utils.handleNanInfValue;
 
 import com.google.common.collect.ImmutableMap;
 import java.util.List;
@@ -161,7 +161,9 @@ void unsupported_aggregation_should_fail() {
 
   @Test
   void nan_value_should_return_null() {
-    assertNull(handleNanValue(Double.NaN));
+    assertNull(handleNanInfValue(Double.NaN));
+    assertNull(handleNanInfValue(Double.NEGATIVE_INFINITY));
+    assertNull(handleNanInfValue(Double.POSITIVE_INFINITY));
   }
 
   @Test