ES|QL: Implement percentiles aggregation for tdigest (elastic#139301)

Author: JonasKunz

server/src/main/java/org/elasticsearch/index/mapper/BlockLoader.java

@@ -688,6 +688,8 @@ Block buildExponentialHistogramBlockDirect(
         );
 
         Block buildTDigestBlockDirect(Block encodedDigests, Block minima, Block maxima, Block sums, Block valueCounts);
+
+        TDigestBuilder tdigestBlockBuilder(int count);
     }
 
     /**

test/framework/src/main/java/org/elasticsearch/index/mapper/TestBlock.java

@@ -553,6 +553,12 @@ public BlockLoader.Block buildTDigestBlockDirect(
 
                 return new TestBlock(values);
             }
+
+            @Override
+            public BlockLoader.TDigestBuilder tdigestBlockBuilder(int count) {
+                // TODO: implement when needed
+                throw new UnsupportedOperationException();
+            }
         };
     }
 

x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/mapper/TDigestBlockLoader.java

@@ -50,7 +50,7 @@ public AllReader reader(LeafReaderContext context) throws IOException {
 
     @Override
     public Builder builder(BlockFactory factory, int expectedCount) {
-        return null;
+        return factory.tdigestBlockBuilder(expectedCount);
     }
 
     static class TDigestReader implements AllReader {
@@ -112,12 +112,12 @@ public Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFilt
 
         @Override
         public void read(int docId, StoredFields storedFields, Builder builder) throws IOException {
-            ExponentialHistogramBuilder histogramBuilder = (ExponentialHistogramBuilder) builder;
+            TDigestBuilder histogramBuilder = (TDigestBuilder) builder;
             minimaReader.read(docId, storedFields, histogramBuilder.minima());
             maximaReader.read(docId, storedFields, histogramBuilder.maxima());
             sumsReader.read(docId, storedFields, histogramBuilder.sums());
             valueCountsReader.read(docId, storedFields, histogramBuilder.valueCounts());
-            encodedDigestReader.read(docId, storedFields, histogramBuilder.encodedHistograms());
+            encodedDigestReader.read(docId, storedFields, histogramBuilder.encodedDigests());
         }
 
         @Override

x-pack/plugin/esql/compute/gen/src/main/java/org/elasticsearch/compute/gen/Methods.java

@@ -317,6 +317,9 @@ public static String getMethod(TypeName elementType) {
         if (elementType.equals(Types.EXPONENTIAL_HISTOGRAM)) {
             return "getExponentialHistogram";
         }
+        if (elementType.equals(Types.TDIGEST)) {
+            return "getTDigestHolder";
+        }
         throw new IllegalArgumentException("unknown get method for [" + elementType + "]");
     }
 

x-pack/plugin/esql/compute/gen/src/main/java/org/elasticsearch/compute/gen/Types.java

@@ -137,6 +137,7 @@ public class Types {
 
     public static final ClassName BYTES_REF = ClassName.get("org.apache.lucene.util", "BytesRef");
     public static final ClassName EXPONENTIAL_HISTOGRAM = ClassName.get("org.elasticsearch.exponentialhistogram", "ExponentialHistogram");
+    public static final ClassName TDIGEST = ClassName.get("org.elasticsearch.compute.data", "TDigestHolder");
 
     public static final ClassName RELEASABLE = ClassName.get("org.elasticsearch.core", "Releasable");
     public static final ClassName RELEASABLES = ClassName.get("org.elasticsearch.core", "Releasables");
@@ -161,7 +162,8 @@ public static TypeDef of(TypeName type, String alias, String block, String vecto
         TypeDef.of(TypeName.FLOAT, "FLOAT", "FloatBlock", "FloatVector", null),
         TypeDef.of(TypeName.DOUBLE, "DOUBLE", "DoubleBlock", "DoubleVector", null),
         TypeDef.of(BYTES_REF, "BYTES_REF", "BytesRefBlock", "BytesRefVector", BYTES_REF),
-        TypeDef.of(EXPONENTIAL_HISTOGRAM, "EXPONENTIAL_HISTOGRAM", "ExponentialHistogramBlock", null, EXPONENTIAL_HISTOGRAM_SCRATCH)
+        TypeDef.of(EXPONENTIAL_HISTOGRAM, "EXPONENTIAL_HISTOGRAM", "ExponentialHistogramBlock", null, EXPONENTIAL_HISTOGRAM_SCRATCH),
+        TypeDef.of(TDIGEST, "TDIGEST", "TDigestBlock", null, null)
     )
         .flatMap(def -> Stream.of(def.type.toString(), def.type + "[]", def.alias).map(alias -> Map.entry(alias, def)))
         .collect(toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));