[Backport 2.19-dev] [SQL/PPL] Fix the count(*) and dc(field) to be capped at MAX_INTEGER #4416

core/src/main/java/org/opensearch/sql/expression/aggregation/AggregatorFunctions.java

@@ -96,7 +96,7 @@ private static DefaultFunctionResolver count() {
                             new FunctionSignature(functionName, Collections.singletonList(type)),
                         type ->
                             (functionProperties, arguments) ->
-                                new CountAggregator(arguments, INTEGER))));
+                                new CountAggregator(arguments, LONG))));
     return functionResolver;
   }
 

core/src/main/java/org/opensearch/sql/expression/aggregation/CountAggregator.java

@@ -44,7 +44,7 @@ public String toString() {
 
   /** Count State. */
   protected static class CountState implements AggregationState {
-    protected int count;
+    protected long count;
 
     CountState() {
       this.count = 0;
@@ -56,7 +56,7 @@ public void count(ExprValue value) {
 
     @Override
     public ExprValue result() {
-      return ExprValueUtils.integerValue(count);
+      return ExprValueUtils.longValue(count);
     }
   }
 

core/src/test/java/org/opensearch/sql/analysis/AnalyzerTest.java

@@ -1283,7 +1283,7 @@ public void named_aggregator_with_condition() {
                 emptyList()),
             DSL.named(
                 "count(string_value) filter(where integer_value > 1)",
-                DSL.ref("count(string_value) filter(where integer_value > 1)", INTEGER))),
+                DSL.ref("count(string_value) filter(where integer_value > 1)", LONG))),
         AstDSL.project(
             AstDSL.agg(
                 AstDSL.relation("schema"),

core/src/test/java/org/opensearch/sql/expression/aggregation/CountAggregatorTest.java

@@ -30,43 +30,43 @@ class CountAggregatorTest extends AggregationTest {
   @Test
   public void count_integer_field_expression() {
     ExprValue result = aggregation(DSL.count(DSL.ref("integer_value", INTEGER)), tuples);
-    assertEquals(4, result.value());
+    assertEquals(4L, result.value());
   }
 
   @Test
   public void count_long_field_expression() {
     ExprValue result = aggregation(DSL.count(DSL.ref("long_value", LONG)), tuples);
-    assertEquals(4, result.value());
+    assertEquals(4L, result.value());
   }
 
   @Test
   public void count_float_field_expression() {
     ExprValue result = aggregation(DSL.count(DSL.ref("float_value", FLOAT)), tuples);
-    assertEquals(4, result.value());
+    assertEquals(4L, result.value());
   }
 
   @Test
   public void count_double_field_expression() {
     ExprValue result = aggregation(DSL.count(DSL.ref("double_value", DOUBLE)), tuples);
-    assertEquals(4, result.value());
+    assertEquals(4L, result.value());
   }
 
   @Test
   public void count_date_field_expression() {
     ExprValue result = aggregation(DSL.count(DSL.ref("date_value", DATE)), tuples);
-    assertEquals(4, result.value());
+    assertEquals(4L, result.value());
   }
 
   @Test
   public void count_timestamp_field_expression() {
     ExprValue result = aggregation(DSL.count(DSL.ref("timestamp_value", TIMESTAMP)), tuples);
-    assertEquals(4, result.value());
+    assertEquals(4L, result.value());
   }
 
   @Test
   public void count_datetime_field_expression() {
     ExprValue result = aggregation(DSL.count(DSL.ref("datetime_value", DATETIME)), tuples);
-    assertEquals(4, result.value());
+    assertEquals(4L, result.value());
   }
 
   @Test
@@ -75,34 +75,33 @@ public void count_arithmetic_expression() {
         aggregation(
             DSL.count(
                 DSL.multiply(
-                    DSL.ref("integer_value", INTEGER),
-                    DSL.literal(ExprValueUtils.integerValue(10)))),
+                    DSL.ref("long_value", LONG), DSL.literal(ExprValueUtils.longValue(10L)))),
             tuples);
-    assertEquals(4, result.value());
+    assertEquals(4L, result.value());
   }
 
   @Test
   public void count_string_field_expression() {
     ExprValue result = aggregation(DSL.count(DSL.ref("string_value", STRING)), tuples);
-    assertEquals(4, result.value());
+    assertEquals(4L, result.value());
   }
 
   @Test
   public void count_boolean_field_expression() {
     ExprValue result = aggregation(DSL.count(DSL.ref("boolean_value", BOOLEAN)), tuples);
-    assertEquals(1, result.value());
+    assertEquals(1L, result.value());
   }
 
   @Test
   public void count_struct_field_expression() {
     ExprValue result = aggregation(DSL.count(DSL.ref("struct_value", STRUCT)), tuples);
-    assertEquals(1, result.value());
+    assertEquals(1L, result.value());
   }
 
   @Test
   public void count_array_field_expression() {
     ExprValue result = aggregation(DSL.count(DSL.ref("array_value", ARRAY)), tuples);
-    assertEquals(1, result.value());
+    assertEquals(1L, result.value());
   }
 
   @Test
@@ -112,14 +111,14 @@ public void filtered_count() {
             DSL.count(DSL.ref("integer_value", INTEGER))
                 .condition(DSL.greater(DSL.ref("integer_value", INTEGER), DSL.literal(1))),
             tuples);
-    assertEquals(3, result.value());
+    assertEquals(3L, result.value());
   }
 
   @Test
   public void distinct_count() {
     ExprValue result =
         aggregation(DSL.distinctCount(DSL.ref("integer_value", INTEGER)), tuples_with_duplicates);
-    assertEquals(3, result.value());
+    assertEquals(3L, result.value());
   }
 
   @Test
@@ -129,47 +128,47 @@ public void filtered_distinct_count() {
             DSL.distinctCount(DSL.ref("integer_value", INTEGER))
                 .condition(DSL.greater(DSL.ref("double_value", DOUBLE), DSL.literal(1d))),
             tuples_with_duplicates);
-    assertEquals(2, result.value());
+    assertEquals(2L, result.value());
   }
 
   @Test
   public void distinct_count_map() {
     ExprValue result =
         aggregation(DSL.distinctCount(DSL.ref("struct_value", STRUCT)), tuples_with_duplicates);
-    assertEquals(3, result.value());
+    assertEquals(3L, result.value());
   }
 
   @Test
   public void distinct_count_array() {
     ExprValue result =
         aggregation(DSL.distinctCount(DSL.ref("array_value", ARRAY)), tuples_with_duplicates);
-    assertEquals(3, result.value());
+    assertEquals(3L, result.value());
   }
 
   @Test
   public void count_with_missing() {
     ExprValue result =
         aggregation(DSL.count(DSL.ref("integer_value", INTEGER)), tuples_with_null_and_missing);
-    assertEquals(2, result.value());
+    assertEquals(2L, result.value());
   }
 
   @Test
   public void count_with_null() {
     ExprValue result =
         aggregation(DSL.count(DSL.ref("double_value", DOUBLE)), tuples_with_null_and_missing);
-    assertEquals(2, result.value());
+    assertEquals(2L, result.value());
   }
 
   @Test
   public void count_star_with_null_and_missing() {
     ExprValue result = aggregation(DSL.count(DSL.literal("*")), tuples_with_null_and_missing);
-    assertEquals(3, result.value());
+    assertEquals(3L, result.value());
   }
 
   @Test
   public void count_literal_with_null_and_missing() {
     ExprValue result = aggregation(DSL.count(DSL.literal(1)), tuples_with_null_and_missing);
-    assertEquals(3, result.value());
+    assertEquals(3L, result.value());
   }
 
   @Test

integ-test/src/test/java/org/opensearch/sql/bwc/SQLBackwardsCompatibilityIT.java

@@ -175,13 +175,21 @@ private void verifySQLQueries(String endpoint) throws IOException {
         executeSQLQuery(
             endpoint,
             "SELECT COUNT(*) FILTER(WHERE age > 35) FROM " + TestsConstants.TEST_INDEX_ACCOUNT);
-    verifySchema(filterResponse, schema("COUNT(*) FILTER(WHERE age > 35)", null, "integer"));
+    // Accept both integer and long types for backwards compatibility
+    String actualType =
+        (String) filterResponse.getJSONArray("schema").getJSONObject(0).query("/type");
+    String expectedType = actualType.equals("integer") ? "integer" : "long";
+    verifySchema(filterResponse, schema("COUNT(*) FILTER(WHERE age > 35)", null, expectedType));
     verifyDataRows(filterResponse, rows(238));
 
     JSONObject aggResponse =
         executeSQLQuery(
             endpoint, "SELECT COUNT(DISTINCT age) FROM " + TestsConstants.TEST_INDEX_ACCOUNT);
-    verifySchema(aggResponse, schema("COUNT(DISTINCT age)", null, "integer"));
+    // Accept both integer and long types for backwards compatibility
+    String actualType2 =
+        (String) aggResponse.getJSONArray("schema").getJSONObject(0).query("/type");
+    String expectedType2 = actualType2.equals("integer") ? "integer" : "long";
+    verifySchema(aggResponse, schema("COUNT(DISTINCT age)", null, expectedType2));
     verifyDataRows(aggResponse, rows(21));
 
     JSONObject groupByResponse =

integ-test/src/test/java/org/opensearch/sql/correctness/runner/resultset/DBResult.java

@@ -34,6 +34,12 @@ public class DBResult {
   /** Possible types for varchar. H2 2.x use CHARACTER VARYING instead of VARCHAR. */
   private static final Set<String> VARCHAR = ImmutableSet.of("CHARACTER VARYING", "VARCHAR");
 
+  /**
+   * Possible types for integer numbers.<br>
+   * Different databases may return INTEGER or BIGINT for count operations.
+   */
+  private static final Set<String> INTEGER_TYPES = ImmutableSet.of("INTEGER", "BIGINT");
+
   /** Database name for display */
   private final String databaseName;
 
@@ -74,6 +80,8 @@ public void addColumn(String name, String type) {
       type = FLOAT_TYPES.toString();
     } else if (VARCHAR.contains(type)) {
       type = "VARCHAR";
+    } else if (INTEGER_TYPES.contains(type)) {
+      type = INTEGER_TYPES.toString();
     }
     schema.add(new Type(StringUtils.toUpper(name), type));
   }

integ-test/src/test/java/org/opensearch/sql/correctness/runner/resultset/Row.java

@@ -10,20 +10,19 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
-import lombok.EqualsAndHashCode;
+import java.util.Objects;
 import lombok.Getter;
 import lombok.ToString;
 
 /** Row in result set. */
-@EqualsAndHashCode
 @ToString
 @Getter
 public class Row implements Comparable<Row> {
 
   private final Collection<Object> values;
 
   public Row() {
-    this(new ArrayList<>()); // values in order by default
+    this(new ArrayList<>());
   }
 
   public Row(Collection<Object> values) {
@@ -37,7 +36,7 @@ public void add(Object value) {
   private Object roundFloatNum(Object value) {
     if (value instanceof Float) {
       BigDecimal decimal = BigDecimal.valueOf((Float) value).setScale(2, RoundingMode.CEILING);
-      value = decimal.doubleValue(); // Convert to double too
+      value = decimal.doubleValue();
     } else if (value instanceof Double) {
       BigDecimal decimal = BigDecimal.valueOf((Double) value).setScale(2, RoundingMode.CEILING);
       value = decimal.doubleValue();
@@ -70,8 +69,54 @@ public int compareTo(Row other) {
         if (result != 0) {
           return result;
         }
-      } // Ignore incomparable field silently?
+      }
     }
     return 0;
   }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (!(o instanceof Row)) return false;
+    Row other = (Row) o;
+    return valuesEqual(this.values, other.values);
+  }
+
+  private boolean valuesEqual(Collection<Object> values1, Collection<Object> values2) {
+    if (values1.size() != values2.size()) return false;
+
+    List<Object> list1 = new ArrayList<>(values1);
+    List<Object> list2 = new ArrayList<>(values2);
+
+    for (int i = 0; i < list1.size(); i++) {
+      if (!isValueEqual(list1.get(i), list2.get(i))) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  private boolean isValueEqual(Object val1, Object val2) {
+    if (Objects.equals(val1, val2)) return true;
+
+    if (isIntegerOrLong(val1) && isIntegerOrLong(val2)) {
+      return ((Number) val1).longValue() == ((Number) val2).longValue();
+    }
+
+    return false;
+  }
+
+  private boolean isIntegerOrLong(Object value) {
+    return value instanceof Integer || value instanceof Long;
+  }
+
+  @Override
+  public int hashCode() {
+
+    List<Object> normalizedValues = new ArrayList<>();
+    for (Object value : values) {
+      normalizedValues.add(value instanceof Integer ? ((Integer) value).longValue() : value);
+    }
+    return normalizedValues.hashCode();
+  }
 }

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

@@ -204,7 +204,7 @@ public void groupByDateShouldPass() {
                 Index.BANK.getName()));
 
     verifySchema(
-        response, schema("birthdate", null, "timestamp"), schema("count(*)", "count", "integer"));
+        response, schema("birthdate", null, "timestamp"), schema("count(*)", "count", "long"));
     verifyDataRows(response, rows("2018-06-23 00:00:00", 1));
   }
 
@@ -220,9 +220,7 @@ public void groupByDateWithAliasShouldPass() {
                 Index.BANK.getName()));
 
     verifySchema(
-        response,
-        schema("birthdate", "birth", "timestamp"),
-        schema("count(*)", "count", "integer"));
+        response, schema("birthdate", "birth", "timestamp"), schema("count(*)", "count", "long"));
     verifyDataRows(response, rows("2018-06-23 00:00:00", 1));
   }