SOLR-18221 Add support for percentile in rollup for streaming expressions (#4504)

Co-authored-by: Eric Pugh <epugh@opensourceconnections.com>

Author: KhushJain

changelog/unreleased/SOLR-18221-support-percentile-in-rollup.yml

@@ -0,0 +1,7 @@
+title: "Support 'percentile' (per) metric in rollup for streaming expressions"
+type: added
+authors:
+  - name: khushjain
+links:
+  - name: SOLR-18221
+    url: https://issues.apache.org/jira/browse/SOLR-18221

solr/core/gradle.lockfile

@@ -34,7 +34,7 @@ com.ibm.icu:icu4j:77.1=jarValidation,testRuntimeClasspath
 com.j256.simplemagic:simplemagic:1.17=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
 com.jayway.jsonpath:json-path:2.9.0=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
 com.microsoft.onnxruntime:onnxruntime:1.26.0=jarValidation,testRuntimeClasspath
-com.tdunning:t-digest:3.3=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
+com.tdunning:t-digest:3.3=apiHelper,compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
 commons-cli:commons-cli:1.11.0=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
 commons-codec:commons-codec:1.21.0=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
 commons-io:commons-io:2.21.0=apiHelper,compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath

solr/solr-ref-guide/modules/query-guide/pages/stream-decorator-reference.adoc

@@ -1448,7 +1448,7 @@ For faster aggregation over low to moderate cardinality fields, the `facet` func
 * `StreamExpression` (Mandatory)
 * `over`: (Mandatory) A list of fields to group by.
 * `metrics`: (Mandatory) The list of metrics to compute.
-Currently supported metrics are `sum(col)`, `avg(col)`, `min(col)`, `max(col)`, `count(*)`, `missing(col)`, `countDist(col)`.
+Currently supported metrics are `sum(col)`, `avg(col)`, `min(col)`, `max(col)`, `count(*)`, `missing(col)`, `countDist(col)`, `per(col, percentile)`.
 
 === rollup Syntax
 
@@ -1467,7 +1467,9 @@ rollup(
    avg(a_f),
    count(*),
    missing(a_i),
-   countDist(a_i)
+   countDist(a_i),
+   per(a_i, 50),
+   per(a_f, 75)
 )
 ----
 

solr/solrj-streaming/build.gradle

@@ -25,6 +25,7 @@ dependencies {
   // declare dependencies we use even though already declared by solrj-core
   implementation libs.slf4j.api
   implementation libs.apache.commons.math3
+  implementation libs.tdunning.tdigest
 
   testImplementation project(':solr:test-framework')
   testImplementation project(':solr:core')

solr/solrj-streaming/gradle.lockfile

@@ -30,7 +30,7 @@ com.google.protobuf:protobuf-java:3.25.8=annotationProcessor,errorprone,jarValid
 com.googlecode.json-simple:json-simple:1.1.1=jarValidation,testRuntimeClasspath
 com.j256.simplemagic:simplemagic:1.17=jarValidation,testRuntimeClasspath
 com.jayway.jsonpath:json-path:2.9.0=jarValidation,testRuntimeClasspath
-com.tdunning:t-digest:3.3=jarValidation,testRuntimeClasspath
+com.tdunning:t-digest:3.3=compileClasspath,jarValidation,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
 commons-cli:commons-cli:1.11.0=jarValidation,testRuntimeClasspath
 commons-codec:commons-codec:1.21.0=jarValidation,testRuntimeClasspath
 commons-io:commons-io:2.21.0=jarValidation,testCompileClasspath,testRuntimeClasspath

solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/metrics/PercentileMetric.java

@@ -16,6 +16,7 @@
  */
 package org.apache.solr.client.solrj.io.stream.metrics;
 
+import com.tdunning.math.stats.AVLTreeDigest;
 import java.io.IOException;
 import java.util.Locale;
 import org.apache.solr.client.solrj.io.Tuple;
@@ -24,27 +25,27 @@
 import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
 
 public class PercentileMetric extends Metric {
-  private long longMax = -Long.MIN_VALUE;
-  private double doubleMax = -Double.MAX_VALUE;
-  private String columnName;
 
-  public PercentileMetric(String columnName, int percentile) {
+  private static final int COMPRESSION = 100;
+
+  private String columnName;
+  private double percentile;
+  private AVLTreeDigest digest;
 
+  public PercentileMetric(String columnName, double percentile) {
     init("per", columnName, percentile);
   }
 
   public PercentileMetric(StreamExpression expression, StreamFactory factory) throws IOException {
-    // grab all parameters out
     String functionName = expression.getFunctionName();
     String columnName = factory.getValueOperand(expression, 0);
-    int percentile = Integer.parseInt(factory.getValueOperand(expression, 1));
+    String percentileStr = factory.getValueOperand(expression, 1);
 
-    // validate expression contains only what we want.
     if (null == columnName) {
       throw new IOException(
           String.format(
               Locale.ROOT,
-              "Invalid expression %s - expected %s(columnName)",
+              "Invalid expression %s - expected %s(columnName, percentile)",
               expression,
               functionName));
     }
@@ -53,39 +54,64 @@ public PercentileMetric(StreamExpression expression, StreamFactory factory) thro
           String.format(Locale.ROOT, "Invalid expression %s - unknown operands found", expression));
     }
 
+    double percentile = Double.parseDouble(percentileStr);
     init(functionName, columnName, percentile);
   }
 
-  private void init(String functionName, String columnName, int percentile) {
+  private void init(String functionName, String columnName, double percentile) {
+    if (percentile < 0 || percentile > 100) {
+      throw new IllegalArgumentException("percentile must be between 0 and 100, got " + percentile);
+    }
     this.columnName = columnName;
+    this.percentile = percentile;
+    this.digest = new AVLTreeDigest(COMPRESSION);
     setFunctionName(functionName);
-    setIdentifier(functionName, "(", columnName, "," + percentile, ")");
+    setIdentifier(functionName, "(", columnName, "," + formatPercentile(percentile), ")");
   }
 
-  @Override
-  public Number getValue() {
-    if (longMax == Long.MIN_VALUE) {
-      return doubleMax;
-    } else {
-      return longMax;
+  private static String formatPercentile(double percentile) {
+    if (percentile == Math.floor(percentile) && !Double.isInfinite(percentile)) {
+      return Integer.toString((int) percentile);
     }
+    return String.valueOf(percentile);
   }
 
   @Override
-  public String[] getColumns() {
-    return new String[] {columnName};
+  public void update(Tuple tuple) {
+    Object o = tuple.get(columnName);
+    if (o instanceof Double d) {
+      digest.add(d);
+    } else if (o instanceof Float f) {
+      digest.add(f.doubleValue());
+    } else if (o instanceof Integer i) {
+      digest.add(i.doubleValue());
+    } else if (o instanceof Long l) {
+      digest.add(l.doubleValue());
+    }
   }
 
   @Override
-  public void update(Tuple tuple) {}
+  public Number getValue() {
+    if (digest.size() == 0) {
+      return null;
+    }
+    return digest.quantile(percentile / 100.0);
+  }
 
   @Override
   public Metric newInstance() {
-    return new MaxMetric(columnName);
+    return new PercentileMetric(columnName, percentile);
+  }
+
+  @Override
+  public String[] getColumns() {
+    return new String[] {columnName};
   }
 
   @Override
   public StreamExpressionParameter toExpression(StreamFactory factory) throws IOException {
-    return new StreamExpression(getFunctionName()).withParameter(columnName);
+    return new StreamExpression(getFunctionName())
+        .withParameter(columnName)
+        .withParameter(formatPercentile(percentile));
   }
 }

solr/solrj-streaming/src/test/org/apache/solr/client/solrj/io/stream/StreamingTest.java

@@ -39,6 +39,7 @@
 import org.apache.solr.client.solrj.io.comp.MultipleFieldComparator;
 import org.apache.solr.client.solrj.io.eq.FieldEqualitor;
 import org.apache.solr.client.solrj.io.ops.GroupOperation;
+import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionParameter;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionParser;
 import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
 import org.apache.solr.client.solrj.io.stream.metrics.Bucket;
@@ -49,6 +50,7 @@
 import org.apache.solr.client.solrj.io.stream.metrics.Metric;
 import org.apache.solr.client.solrj.io.stream.metrics.MinMetric;
 import org.apache.solr.client.solrj.io.stream.metrics.MissingMetric;
+import org.apache.solr.client.solrj.io.stream.metrics.PercentileMetric;
 import org.apache.solr.client.solrj.io.stream.metrics.SumMetric;
 import org.apache.solr.client.solrj.request.CollectionAdminRequest;
 import org.apache.solr.client.solrj.request.UpdateRequest;
@@ -1855,6 +1857,114 @@ public void testRollupStream() throws Exception {
     }
   }
 
+  @Test
+  public void testPercentileRollupStream() throws Exception {
+
+    helloDocsUpdateRequest.commit(cluster.getSolrClient(), COLLECTIONORALIAS);
+    StreamContext streamContext = new StreamContext();
+    SolrClientCache solrClientCache = new SolrClientCache();
+    streamContext.setSolrClientCache(solrClientCache);
+
+    try {
+      // Test percentile metrics in RollupStream
+      SolrParams sParamsA = params("q", "*:*", "fl", "a_s,a_i,a_f", "sort", "a_s asc");
+      CloudSolrStream stream = new CloudSolrStream(solrConnection, COLLECTIONORALIAS, sParamsA);
+
+      Bucket[] buckets = {new Bucket("a_s")};
+
+      Metric[] metrics = {
+        new PercentileMetric("a_i", 50),
+        new PercentileMetric("a_f", 50),
+        new PercentileMetric("a_i", 0),
+        new PercentileMetric("a_i", 100),
+        new PercentileMetric("a_i", 99.9),
+        new MinMetric("a_i"),
+        new MaxMetric("a_i"),
+      };
+
+      RollupStream rollupStream = new RollupStream(stream, buckets, metrics);
+      rollupStream.setStreamContext(streamContext);
+      List<Tuple> tuples = getTuples(rollupStream);
+
+      assertEquals(3, tuples.size());
+
+      // hello0: a_i = [0, 1, 2, 14], a_f = [1, 2, 5, 10]
+      Tuple tuple = tuples.get(0);
+      assertEquals("hello0", tuple.getString("a_s"));
+      Double peri50 = tuple.getDouble("per(a_i,50)");
+      Double perf50 = tuple.getDouble("per(a_f,50)");
+      Double peri0 = tuple.getDouble("per(a_i,0)");
+      Double peri100 = tuple.getDouble("per(a_i,100)");
+      Double peri999 = tuple.getDouble("per(a_i,99.9)");
+      Double mini = tuple.getDouble("min(a_i)");
+      Double maxi = tuple.getDouble("max(a_i)");
+
+      assertNotNull(peri50);
+      assertNotNull(perf50);
+      // 0th percentile should approximate min
+      assertEquals(mini, peri0, 0.5);
+      // 100th percentile should approximate max
+      assertEquals(maxi, peri100, 0.5);
+      // 99.9th percentile should be close to max
+      assertNotNull(peri999);
+
+      // hello3: a_i = [3, 10, 12, 13], a_f = [3, 6, 8, 9]
+      tuple = tuples.get(1);
+      assertEquals("hello3", tuple.getString("a_s"));
+      peri50 = tuple.getDouble("per(a_i,50)");
+      perf50 = tuple.getDouble("per(a_f,50)");
+      peri0 = tuple.getDouble("per(a_i,0)");
+      peri100 = tuple.getDouble("per(a_i,100)");
+      mini = tuple.getDouble("min(a_i)");
+      maxi = tuple.getDouble("max(a_i)");
+
+      assertNotNull(peri50);
+      assertNotNull(perf50);
+      assertEquals(mini, peri0, 0.5);
+      assertEquals(maxi, peri100, 0.5);
+
+      // hello4: a_i = [4, 11], a_f = [4, 7]
+      tuple = tuples.get(2);
+      assertEquals("hello4", tuple.getString("a_s"));
+      peri50 = tuple.getDouble("per(a_i,50)");
+      perf50 = tuple.getDouble("per(a_f,50)");
+      peri0 = tuple.getDouble("per(a_i,0)");
+      peri100 = tuple.getDouble("per(a_i,100)");
+      mini = tuple.getDouble("min(a_i)");
+      maxi = tuple.getDouble("max(a_i)");
+
+      assertNotNull(peri50);
+      assertNotNull(perf50);
+      assertEquals(mini, peri0, 0.5);
+      assertEquals(maxi, peri100, 0.5);
+
+      // Test toExpression round-trip
+      PercentileMetric pm = new PercentileMetric("a_i", 50);
+      StreamExpressionParameter expr = pm.toExpression(streamFactory);
+      assertEquals("per(a_i,50)", expr.toString());
+
+      pm = new PercentileMetric("a_i", 99.9);
+      expr = pm.toExpression(streamFactory);
+      assertEquals("per(a_i,99.9)", expr.toString());
+
+      // Test HashRollupStream with percentile
+      stream = new CloudSolrStream(solrConnection, COLLECTIONORALIAS, sParamsA);
+      Metric[] hashMetrics = {new PercentileMetric("a_i", 50), new PercentileMetric("a_f", 50)};
+
+      HashRollupStream hashRollupStream = new HashRollupStream(stream, buckets, hashMetrics);
+      hashRollupStream.setStreamContext(streamContext);
+      tuples = getTuples(hashRollupStream);
+
+      assertEquals(3, tuples.size());
+      for (Tuple t : tuples) {
+        assertNotNull(t.getDouble("per(a_i,50)"));
+        assertNotNull(t.getDouble("per(a_f,50)"));
+      }
+    } finally {
+      solrClientCache.close();
+    }
+  }
+
   @Test
   public void testDaemonTopicStream() throws Exception {
     Assume.assumeTrue(!useAlias);