Add values stats function with UDAF

common/src/main/java/org/opensearch/sql/common/setting/Settings.java

@@ -30,6 +30,7 @@ public enum Key {
     PATTERN_MAX_SAMPLE_COUNT("plugins.ppl.pattern.max.sample.count"),
     PATTERN_BUFFER_LIMIT("plugins.ppl.pattern.buffer.limit"),
     PPL_REX_MAX_MATCH_LIMIT("plugins.ppl.rex.max_match.limit"),
+    PPL_VALUES_MAX_LIMIT("plugins.ppl.values.max.limit"),
     PPL_SYNTAX_LEGACY_PREFERRED("plugins.ppl.syntax.legacy.preferred"),
 
     /** Enable Calcite as execution engine */

core/src/main/java/org/opensearch/sql/calcite/udf/udaf/ListAggFunction.java

@@ -21,9 +21,8 @@
  *   <li>Order of values in the result is non-deterministic
  * </ul>
  *
- * <p>Note: Similar to the TAKE function, LIST does not guarantee any specific order of values in
- * the result array. The order may vary between executions and depends on the underlying query
- * execution plan and optimizations.
+ * <p>LIST does not guarantee any specific order of values in the result array. The order may vary
+ * between executions and depends on the underlying query execution plan and optimizations.
  */
 public class ListAggFunction implements UserDefinedAggFunction<ListAggFunction.ListAccumulator> {
 

core/src/main/java/org/opensearch/sql/calcite/udf/udaf/ValuesAggFunction.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.calcite.udf.udaf;
+
+import java.util.ArrayList;
+import java.util.Set;
+import java.util.TreeSet;
+import org.opensearch.sql.calcite.udf.UserDefinedAggFunction;
+
+/**
+ * VALUES aggregate function implementation. Returns distinct values from a field in lexicographical
+ * order as a multivalue field.
+ *
+ * <p>Behavior:
+ *
+ * <ul>
+ *   <li>Returns unique values only (no duplicates)
+ *   <li>Values are sorted in lexicographical order
+ *   <li>Processes field values as strings (casts all inputs to strings)
+ *   <li>Configurable limit via plugins.ppl.values.max.limit setting (0 = unlimited)
+ *   <li>Supports only scalar data types (rejects STRUCT/ARRAY types)
+ *   <li>Implementation uses TreeSet for automatic sorting and deduplication
+ * </ul>
+ */
+public class ValuesAggFunction
+    implements UserDefinedAggFunction<ValuesAggFunction.ValuesAccumulator> {
+
+  @Override
+  public ValuesAccumulator init() {
+    return new ValuesAccumulator();
+  }
+
+  @Override
+  public Object result(ValuesAccumulator accumulator) {
+    return accumulator.value();
+  }
+
+  @Override
+  public ValuesAccumulator add(ValuesAccumulator acc, Object... values) {
+    // Handle case where no values are passed
+    if (values == null || values.length == 0) {
+      return acc;
+    }
+
+    Object value = values[0];
+
+    // Get limit from second argument (passed from AST)
+    int limit = 0; // Default to unlimited
+    if (values.length > 1 && values[1] != null) {
+      limit = (Integer) values[1];
+    }
+
+    // Filter out null values and check limit
+    if (value != null && (limit == 0 || acc.size() < limit)) {
+      // Convert value to string
+      String stringValue = String.valueOf(value);
+      acc.add(stringValue, limit);
+    }
+
+    return acc;
+  }
+
+  public static class ValuesAccumulator implements Accumulator {
+    private final Set<String> values;
+
+    public ValuesAccumulator() {
+      this.values = new TreeSet<>(); // TreeSet maintains sorted order and uniqueness
+    }
+
+    @Override
+    public Object value(Object... argList) {
+      return new ArrayList<>(values); // Return List<String> to match expected type
+    }
+
+    public void add(String value, int limit) {
+      if (limit == 0 || values.size() < limit) {
+        values.add(value);
+      }
+    }
+
+    public int size() {
+      return values.size();
+    }
+  }
+}

core/src/main/java/org/opensearch/sql/calcite/utils/PPLOperandTypes.java

@@ -22,6 +22,45 @@ public class PPLOperandTypes {
   // This class is not meant to be instantiated.
   private PPLOperandTypes() {}
 
+  /** List of all scalar type signatures (single parameter each) */
+  private static final java.util.List<java.util.List<org.opensearch.sql.data.type.ExprType>>
+      SCALAR_TYPES =
+          java.util.List.of(
+              // Numeric types
+              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.BYTE),
+              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.SHORT),
+              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.INTEGER),
+              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.LONG),
+              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.FLOAT),
+              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.DOUBLE),
+              // String type
+              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.STRING),
+              // Boolean type
+              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.BOOLEAN),
+              // Temporal types
+              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.DATE),
+              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.TIME),
+              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.TIMESTAMP),
+              // Special scalar types
+              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.IP),
+              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.BINARY));
+
+  /** Helper method to create scalar types with optional integer parameter */
+  private static java.util.List<java.util.List<org.opensearch.sql.data.type.ExprType>>
+      createScalarWithOptionalInteger() {
+    java.util.List<java.util.List<org.opensearch.sql.data.type.ExprType>> result =
+        new java.util.ArrayList<>(SCALAR_TYPES);
+
+    // Add scalar + integer combinations
+    SCALAR_TYPES.forEach(
+        scalarType ->
+            result.add(
+                java.util.List.of(
+                    scalarType.get(0), org.opensearch.sql.data.type.ExprCoreType.INTEGER)));
+
+    return result;
+  }
+
   public static final UDFOperandMetadata NONE = UDFOperandMetadata.wrap(OperandTypes.family());
   public static final UDFOperandMetadata OPTIONAL_ANY =
       UDFOperandMetadata.wrap(
@@ -200,25 +239,12 @@ private PPLOperandTypes() {}
    * booleans, datetime types, and special scalar types like IP and BINARY. Excludes complex types
    * like arrays, structs, and maps.
    */
-  public static final UDFOperandMetadata ANY_SCALAR =
-      UDFOperandMetadata.wrapUDT(
-          java.util.List.of(
-              // Numeric types
-              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.BYTE),
-              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.SHORT),
-              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.INTEGER),
-              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.LONG),
-              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.FLOAT),
-              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.DOUBLE),
-              // String type
-              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.STRING),
-              // Boolean type
-              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.BOOLEAN),
-              // Temporal types
-              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.DATE),
-              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.TIME),
-              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.TIMESTAMP),
-              // Special scalar types
-              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.IP),
-              java.util.List.of(org.opensearch.sql.data.type.ExprCoreType.BINARY)));
+  public static final UDFOperandMetadata ANY_SCALAR = UDFOperandMetadata.wrapUDT(SCALAR_TYPES);
+
+  /**
+   * Operand type checker that accepts any scalar type with an optional integer argument. This is
+   * used for aggregation functions that take a field and an optional limit/size parameter.
+   */
+  public static final UDFOperandMetadata ANY_SCALAR_OPTIONAL_INTEGER =
+      UDFOperandMetadata.wrapUDT(createScalarWithOptionalInteger());
 }

core/src/main/java/org/opensearch/sql/expression/function/BuiltinFunctionName.java

@@ -207,6 +207,7 @@ public enum BuiltinFunctionName {
 
   // Multivalue aggregation function
   LIST(FunctionName.of("list")),
+  VALUES(FunctionName.of("values")),
   // Not always an aggregation query
   NESTED(FunctionName.of("nested")),
   // Document order aggregation functions
@@ -364,6 +365,7 @@ public enum BuiltinFunctionName {
           .put("latest", BuiltinFunctionName.LATEST)
           .put("distinct_count_approx", BuiltinFunctionName.DISTINCT_COUNT_APPROX)
           .put("list", BuiltinFunctionName.LIST)
+          .put("values", BuiltinFunctionName.VALUES)
           .put("pattern", BuiltinFunctionName.INTERNAL_PATTERN)
           .put("first", BuiltinFunctionName.FIRST)
           .put("last", BuiltinFunctionName.LAST)

core/src/main/java/org/opensearch/sql/expression/function/PPLBuiltinOperators.java

@@ -36,6 +36,7 @@
 import org.opensearch.sql.calcite.udf.udaf.NullableSqlAvgAggFunction;
 import org.opensearch.sql.calcite.udf.udaf.PercentileApproxFunction;
 import org.opensearch.sql.calcite.udf.udaf.TakeAggFunction;
+import org.opensearch.sql.calcite.udf.udaf.ValuesAggFunction;
 import org.opensearch.sql.calcite.utils.PPLOperandTypes;
 import org.opensearch.sql.calcite.utils.PPLReturnTypes;
 import org.opensearch.sql.calcite.utils.UserDefinedFunctionUtils;
@@ -450,6 +451,12 @@ public class PPLBuiltinOperators extends ReflectiveSqlOperatorTable {
   public static final SqlAggFunction LIST =
       createUserDefinedAggFunction(
           ListAggFunction.class, "LIST", PPLReturnTypes.STRING_ARRAY, PPLOperandTypes.ANY_SCALAR);
+  public static final SqlAggFunction VALUES =
+      createUserDefinedAggFunction(
+          ValuesAggFunction.class,
+          "VALUES",
+          PPLReturnTypes.STRING_ARRAY,
+          PPLOperandTypes.ANY_SCALAR_OPTIONAL_INTEGER);
 
   public static final SqlOperator ENHANCED_COALESCE =
       new EnhancedCoalesceFunction().toUDF("COALESCE");

core/src/main/java/org/opensearch/sql/expression/function/PPLFuncImpTable.java

@@ -217,6 +217,7 @@
 import static org.opensearch.sql.expression.function.BuiltinFunctionName.UTC_DATE;
 import static org.opensearch.sql.expression.function.BuiltinFunctionName.UTC_TIME;
 import static org.opensearch.sql.expression.function.BuiltinFunctionName.UTC_TIMESTAMP;
+import static org.opensearch.sql.expression.function.BuiltinFunctionName.VALUES;
 import static org.opensearch.sql.expression.function.BuiltinFunctionName.VARPOP;
 import static org.opensearch.sql.expression.function.BuiltinFunctionName.VARSAMP;
 import static org.opensearch.sql.expression.function.BuiltinFunctionName.WEEK;
@@ -1120,6 +1121,7 @@ void populate() {
       registerOperator(TAKE, PPLBuiltinOperators.TAKE);
       registerOperator(INTERNAL_PATTERN, PPLBuiltinOperators.INTERNAL_PATTERN);
       registerOperator(LIST, PPLBuiltinOperators.LIST);
+      registerOperator(VALUES, PPLBuiltinOperators.VALUES);
 
       register(
           AVG,

docs/user/ppl/admin/settings.rst

@@ -226,3 +226,85 @@ PPL query::
         }
       }
     }
+
+plugins.ppl.values.max.limit
+============================
+
+Description
+-----------
+
+This setting controls the maximum number of unique values that the ``VALUES`` aggregation function can return. When set to 0 (the default), there is no limit on the number of unique values returned. When set to a positive integer, the function will return at most that many unique values.
+
+1. The default value is 0 (unlimited).
+2. This setting is node scope.
+3. This setting can be updated dynamically.
+
+The ``VALUES`` function collects all unique values from a field and returns them in lexicographical order. This setting helps manage memory usage by limiting the number of values collected.
+
+Example 1
+---------
+
+Set the limit to 1000 unique values:
+
+PPL query::
+
+    sh$ curl -sS -H 'Content-Type: application/json' \
+    ... -X PUT localhost:9200/_plugins/_query/settings \
+    ... -d '{"transient" : {"plugins.ppl.values.max.limit" : "1000"}}'
+    {
+      "acknowledged": true,
+      "persistent": {},
+      "transient": {
+        "plugins": {
+          "ppl": {
+            "values": {
+              "max": {
+                "limit": "1000"
+              }
+            }
+          }
+        }
+      }
+    }
+
+Example 2
+---------
+
+Reset to default (unlimited) by setting to null:
+
+PPL query::
+
+    sh$ curl -sS -H 'Content-Type: application/json' \
+    ... -X PUT localhost:9200/_plugins/_query/settings \
+    ... -d '{"transient" : {"plugins.ppl.values.max.limit" : null}}'
+    {
+      "acknowledged": true,
+      "persistent": {},
+      "transient": {}
+    }
+
+Example 3
+---------
+
+Set to 0 explicitly for unlimited values:
+
+PPL query::
+
+    sh$ curl -sS -H 'Content-Type: application/json' \
+    ... -X PUT localhost:9200/_plugins/_query/settings \
+    ... -d '{"transient" : {"plugins.ppl.values.max.limit" : "0"}}'
+    {
+      "acknowledged": true,
+      "persistent": {},
+      "transient": {
+        "plugins": {
+          "ppl": {
+            "values": {
+              "max": {
+                "limit": "0"
+              }
+            }
+          }
+        }
+      }
+    }