Upgrade t-digest from 3.2 to 3.3 with error rate fix

LICENSE-binary

@@ -278,7 +278,7 @@ com.squareup.wire:wire-runtime-jvm:5.1.0
 com.squareup.wire:wire-schema-jvm:5.1.0
 com.squareup:javapoet:1.13.0
 com.squareup:kotlinpoet-jvm:1.18.1
-com.tdunning:t-digest:3.2
+com.tdunning:t-digest:3.3
 com.typesafe.scala-logging:scala-logging_2.13:3.9.5
 com.uber:h3:4.4.0
 com.yammer.metrics:metrics-core:2.2.0

pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileSmartTDigestAggregationFunctionTest.java

@@ -34,39 +34,26 @@ public String callStr(String column, int percent) {
       return "PERCENTILESMARTTDIGEST(" + column + ", " + percent + ", 'THRESHOLD=1')";
     }
 
+    // t-digest 3.3 changed interpolation for small datasets: values snap to integers
+    // instead of interpolating between adjacent values (e.g., p10 returns 1.0 not 0.5)
     @Override
     String expectedAggrWithNull10(Scenario scenario) {
-      return "0.5";
+      return "1.0";
     }
 
     @Override
     String expectedAggrWithNull30(Scenario scenario) {
-      return "2.5";
+      return "3.0";
     }
 
     @Override
     String expectedAggrWithNull50(Scenario scenario) {
-      return "4.5";
+      return "5.0";
     }
 
     @Override
     String expectedAggrWithNull70(Scenario scenario) {
-      return "6.5";
-    }
-
-    @Override
-    String expectedAggrWithoutNull55(Scenario scenario) {
-      switch (scenario.getDataType()) {
-        case INT:
-          return "-6.442450943999939E8";
-        case LONG:
-          return "-2.7670116110564065E18";
-        case FLOAT:
-        case DOUBLE:
-          return "-Infinity";
-        default:
-          throw new IllegalArgumentException("Unsupported datatype " + scenario.getDataType());
-      }
+      return "7.0";
     }
 
     @Override
@@ -76,7 +63,7 @@ String expectedAggrWithoutNull75(Scenario scenario) {
 
     @Override
     String expectedAggrWithoutNull90(Scenario scenario) {
-      return "7.100000000000001";
+      return "7.0";
     }
   }
 }

pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestMVAggregationFunctionTest.java

@@ -41,9 +41,9 @@ public void testAggregationMV() {
         .andOnSecondInstance(
             new Object[]{"6.0;7.0;8.0;9.0;10.0"}
         )
-        // All values: 1-10, p50 should be around 5
+        // All values: 1-10, p50 (t-digest approximate)
         .whenQuery("select percentiletdigest(mv, 50) from testTable")
-        .thenResultIs("DOUBLE", "5.5");
+        .thenResultIs("DOUBLE", "6.0");
   }
 
   @Test
@@ -66,7 +66,7 @@ public void testAggregationMVGroupBySV() {
         )
         .whenQuery("select sv, percentiletdigest(mv, 50) from testTable group by sv order by sv")
         .thenResultIs("STRING | DOUBLE",
-            "k1 | 5.5",   // values: 1-10, p50 ~= 5.5
+            "k1 | 6.0",   // values: 1-10, p50 (t-digest approximate)
             "k2 | 30.0"); // values: 10, 20, 30, 40, 50, p50 ~= 30
   }
 
@@ -89,7 +89,7 @@ public void testAggregationMVGroupByMV() {
         )
         .whenQuery("select tags, percentiletdigest(nums, 50) from testTable group by tags order by tags")
         .thenResultIs("STRING | DOUBLE",
-            "tag1 | 3.5",  // nums: 1, 2, 3, 4, 5, 6, p50 ~= 3.5
-            "tag2 | 3.5"); // nums: 1, 2, 3, 4, 5, 6, p50 ~= 3.5
+            "tag1 | 4.0",  // nums: 1, 2, 3, 4, 5, 6, p50 (t-digest approximate)
+            "tag2 | 4.0"); // nums: 1, 2, 3, 4, 5, 6, p50 (t-digest approximate)
   }
 }

pinot-core/src/test/java/org/apache/pinot/core/startree/v2/PreAggregatedPercentileTDigestStarTreeV2Test.java

@@ -30,8 +30,13 @@
 
 
 public class PreAggregatedPercentileTDigestStarTreeV2Test extends BaseStarTreeV2Test<Object, TDigest> {
-  // Use non-default compression
-  private static final double COMPRESSION = 50;
+  // Use high compression to keep star-tree vs non-star-tree quantile divergence within 0.5%.
+  // t-digest 3.3 changed centroid management (unit-weight first/last centroids, stricter tail interpolation),
+  // which increases merge-order sensitivity. The star-tree path does multi-level serialize/deserialize/merge
+  // while the non-star-tree path merges sequentially, causing quantile divergence at low compression values.
+  // Experimentally verified: compression >= 750 keeps error < 0.5% across 10 randomized runs.
+  private static final double COMPRESSION = 750;
+  private static final double MAX_ERROR = 0.005;
   private static final int MAX_VALUE = 10000;
 
   @Override
@@ -54,7 +59,7 @@ Object getRandomRawValue(Random random) {
 
   @Override
   void assertAggregatedValue(TDigest starTreeResult, TDigest nonStarTreeResult) {
-    double delta = MAX_VALUE * 0.05;
+    double delta = MAX_VALUE * MAX_ERROR;
     for (int i = 0; i <= 100; i++) {
       assertEquals(starTreeResult.quantile(i / 100.0), nonStarTreeResult.quantile(i / 100.0), delta);
     }

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java

@@ -29,8 +29,11 @@
 public class PercentileTDigestValueAggregator implements ValueAggregator<Object, TDigest> {
   public static final DataType AGGREGATED_VALUE_TYPE = DataType.BYTES;
 
-  // TODO: This is copied from PercentileTDigestAggregationFunction.
-  public static final int DEFAULT_TDIGEST_COMPRESSION = 100;
+  // NOTE: This is intentionally higher than the query-time default (100) in PercentileTDigestAggregationFunction.
+  // t-digest 3.3 has higher merge-order sensitivity, and star-tree building involves multi-level merge with
+  // intermediate serialization/deserialization. At compression=100, this causes ~0.35% error (vs 0.06% in 3.2).
+  // Compression=500 brings error back to ~0.09%, comparable to 3.2's accuracy at compression=100.
+  public static final int DEFAULT_TDIGEST_COMPRESSION = 500;
   private final int _compressionFactor;
   private int _maxByteSize;
 

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/StarTreeBuilderUtils.java

@@ -34,6 +34,7 @@
 import org.apache.commons.io.FileUtils;
 import org.apache.pinot.common.request.Literal;
 import org.apache.pinot.common.request.context.ExpressionContext;
+import org.apache.pinot.segment.local.aggregator.PercentileTDigestValueAggregator;
 import org.apache.pinot.segment.local.startree.v2.builder.StarTreeV2BuilderConfig;
 import org.apache.pinot.segment.spi.AggregationFunctionType;
 import org.apache.pinot.segment.spi.Constants;
@@ -395,6 +396,12 @@ public static List<ExpressionContext> expressionContextFromFunctionParameters(
           expressionContexts.add(ExpressionContext.forLiteral(
               Literal.intValue(Integer.parseInt(String.valueOf(
                   functionParameters.get(Constants.PERCENTILETDIGEST_COMPRESSION_FACTOR_KEY))))));
+        } else {
+          // Always inject the default compression so it is stored in star-tree metadata.
+          // This allows canUseStarTree() to distinguish old segments (no metadata, built with compression=100)
+          // from new segments (metadata present, built with the current default).
+          expressionContexts.add(ExpressionContext.forLiteral(
+              Literal.intValue(PercentileTDigestValueAggregator.DEFAULT_TDIGEST_COMPRESSION)));
         }
         break;
       }

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/StarTreeV2BuilderConfig.java

@@ -32,8 +32,10 @@
 import org.apache.commons.configuration2.Configuration;
 import org.apache.commons.lang3.builder.ToStringBuilder;
 import org.apache.commons.lang3.builder.ToStringStyle;
+import org.apache.pinot.segment.local.aggregator.PercentileTDigestValueAggregator;
 import org.apache.pinot.segment.spi.AggregationFunctionType;
 import org.apache.pinot.segment.spi.ColumnMetadata;
+import org.apache.pinot.segment.spi.Constants;
 import org.apache.pinot.segment.spi.SegmentMetadata;
 import org.apache.pinot.segment.spi.index.startree.AggregationFunctionColumnPair;
 import org.apache.pinot.segment.spi.index.startree.AggregationSpec;
@@ -80,7 +82,8 @@ public static StarTreeV2BuilderConfig fromIndexConfig(StarTreeIndexConfig indexC
             AggregationFunctionColumnPair.resolveToStoredType(aggregationFunctionColumnPair);
         // If there is already an equivalent functionColumnPair in the map, do not load another.
         // This prevents the duplication of the aggregation when the StarTree is constructed.
-        aggregationSpecs.putIfAbsent(storedType, AggregationSpec.DEFAULT);
+        aggregationSpecs.putIfAbsent(storedType,
+            getDefaultAggregationSpec(aggregationFunctionColumnPair.getFunctionType()));
       }
     }
     if (indexConfig.getAggregationConfigs() != null) {
@@ -323,4 +326,22 @@ public String toString() {
         .append("skipStarNodeCreation", _skipStarNodeCreationForDimensions)
         .append("aggregationSpecs", _aggregationSpecs).append("maxLeafRecords", _maxLeafRecords).toString();
   }
+
+  /**
+   * Returns the default {@link AggregationSpec} for the given function type. For PERCENTILETDIGEST and
+   * PERCENTILERAWTDIGEST, the default compression factor is explicitly included in the function parameters so that
+   * it is persisted in star-tree metadata. This allows {@code canUseStarTree()} to distinguish old segments (built
+   * with the previous default of 100, which have no compression metadata) from new segments.
+   */
+  private static AggregationSpec getDefaultAggregationSpec(AggregationFunctionType functionType) {
+    switch (functionType) {
+      case PERCENTILETDIGEST:
+      case PERCENTILERAWTDIGEST:
+        return new AggregationSpec(null, null, null, null, null,
+            Map.of(Constants.PERCENTILETDIGEST_COMPRESSION_FACTOR_KEY,
+                PercentileTDigestValueAggregator.DEFAULT_TDIGEST_COMPRESSION));
+      default:
+        return AggregationSpec.DEFAULT;
+    }
+  }
 }