opensearch-project · sachin-27 · Jul 21, 2026 · Jul 22, 2026 · Jul 24, 2026 · Jul 29, 2026
@@ -40,6 +40,8 @@ public enum AggregateFunction {
     PERCENTILE_CONT(Type.STATE_EXPANDING, SqlKind.PERCENTILE_CONT),
     PERCENTILE_DISC(Type.STATE_EXPANDING, SqlKind.PERCENTILE_DISC),
     PERCENTILE_APPROX(Type.STATE_EXPANDING, SqlKind.OTHER),
+    /** Explicit-compression percentile: {@code (field, percent, centroids)}. */
+    PERCENTILE_APPROX_N(Type.STATE_EXPANDING, SqlKind.OTHER),
     COLLECT(Type.STATE_EXPANDING, SqlKind.COLLECT),
     LISTAGG(Type.STATE_EXPANDING, SqlKind.LISTAGG),
 

@@ -475,6 +475,7 @@ public class DataFusionAnalyticsBackendPlugin implements AnalyticsSearchBackendP
         AggregateFunction.AVG,
         AggregateFunction.APPROX_COUNT_DISTINCT,
         AggregateFunction.PERCENTILE_APPROX,
+        AggregateFunction.PERCENTILE_APPROX_N,
         AggregateFunction.TAKE,
         AggregateFunction.FIRST,
         AggregateFunction.LAST,

@@ -38,6 +38,7 @@
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.apache.calcite.sql.type.OperandTypes;
 import org.apache.calcite.sql.type.ReturnTypes;
+import org.apache.calcite.sql.type.SqlTypeFamily;
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.calcite.sql.type.SqlTypeTransforms;
 import org.apache.calcite.util.ImmutableBitSet;
@@ -381,6 +382,23 @@ public RexNode normaliseLiteralArg(int argIndex, RexLiteral lit, RexBuilder rexB
         }
     };
 
+    /**
+     * Explicit-compression percentile (DSL executor): {@code PERCENTILE_APPROX_N(field,
+     * percent, centroids)} → {@code approx_percentile_cont}. Percent rescaling delegates to
+     * {@link #LOCAL_PERCENTILE_APPROX_OP}; the centroids literal passes through unrescaled.
+     */
+    static final LocalAggOp LOCAL_PERCENTILE_APPROX_N_OP = new LocalAggOp(
+        "approx_percentile_cont",
+        SqlKind.OTHER_FUNCTION,
+        ReturnTypes.ARG0.andThen(SqlTypeTransforms.FORCE_NULLABLE),
+        OperandTypes.family(SqlTypeFamily.ANY, SqlTypeFamily.ANY, SqlTypeFamily.ANY)
+    ) {
+        @Override
+        public RexNode normaliseLiteralArg(int argIndex, RexLiteral lit, RexBuilder rexBuilder, RelDataTypeFactory typeFactory) {
+            return LOCAL_PERCENTILE_APPROX_OP.normaliseLiteralArg(argIndex, lit, rexBuilder, typeFactory);
+        }
+    };
+
     /** BRAIN window stub for {@code patterns ... method=BRAIN mode=label}. */
     static final SqlAggFunction LOCAL_INTERNAL_PATTERN_WINDOW_OP = new SqlAggFunction(
         "internal_pattern",
@@ -440,6 +458,7 @@ public RexNode normaliseLiteralArg(int argIndex, RexLiteral lit, RexBuilder rexB
         FunctionMappings.s(LOCAL_LIST_MERGE_OP, "list_merge"),
         FunctionMappings.s(LOCAL_LIST_MERGE_DISTINCT_OP, "list_merge_distinct"),
         FunctionMappings.s(LOCAL_PERCENTILE_APPROX_OP, "approx_percentile_cont"),
+        FunctionMappings.s(LOCAL_PERCENTILE_APPROX_N_OP, "approx_percentile_cont"),
         FunctionMappings.s(LOCAL_INTERNAL_PATTERN_OP, "internal_pattern"),
         FunctionMappings.s(LOCAL_OS_COUNT_DISTINCT_OP, "os_count_distinct")
     );

@@ -30,8 +30,9 @@
 import java.util.Set;
 
 /**
- * Rewrites PPL state-expanding aggregates (TAKE / FIRST / LAST / LIST / VALUES / PATTERN /
- * PERCENTILE_APPROX) onto local stubs the substrait emitter binds via
+ * Rewrites state-expanding aggregates (TAKE / FIRST / LAST / LIST / VALUES / PATTERN /
+ * PERCENTILE_APPROX / PERCENTILE_APPROX_N — emitted by PPL and, for the percentile ops,
+ * by the DSL query executor) onto local stubs the substrait emitter binds via
  * {@link DataFusionFragmentConvertor}'s ADDITIONAL_AGGREGATE_SIGS. Also normalises any
  * RexLiteral{SqlTypeName.SYMBOL} in upstream Projects to VARCHAR — isthmus's
  * LiteralConverter rejects unregistered Enum classes, and PPL's percentile_approx /
@@ -186,14 +187,18 @@ private static AggregateCall rewriteCall(Aggregate agg, AggregateCall call) {
                 targetOp = DataFusionFragmentConvertor.LOCAL_INTERNAL_PATTERN_OP;
                 explicitReturnType = internalPatternReturnType(agg.getCluster().getTypeFactory());
             }
-            case "PERCENTILE_APPROX" -> {
-                // Trim the PPL type-flag arg; the substrait emit-time literal-arg normaliser
-                // (DataFusionFragmentConvertor#normaliseLiteralArg) rescales the percentile.
-                if (call.getArgList().size() < 3) {
+            case "PERCENTILE_APPROX", "PERCENTILE_APPROX_N" -> {
+                // PERCENTILE_APPROX: (field, percent) + optional PPL type-flag arg to trim.
+                // PERCENTILE_APPROX_N (DSL executor): (field, percent, centroids), kept whole.
+                boolean withCentroids = "PERCENTILE_APPROX_N".equals(call.getAggregation().getName());
+                int argCount = withCentroids ? 3 : 2;
+                if (call.getArgList().size() < argCount) {
                     return call;
                 }
-                targetOp = DataFusionFragmentConvertor.LOCAL_PERCENTILE_APPROX_OP;
-                List<Integer> trimmedArgList = new ArrayList<>(call.getArgList().subList(0, 2));
+                targetOp = withCentroids
+                    ? DataFusionFragmentConvertor.LOCAL_PERCENTILE_APPROX_N_OP
+                    : DataFusionFragmentConvertor.LOCAL_PERCENTILE_APPROX_OP;
+                List<Integer> trimmedArgList = new ArrayList<>(call.getArgList().subList(0, argCount));
                 RelDataType arg0Type = agg.getInput().getRowType().getFieldList().get(call.getArgList().get(0)).getType();
                 RelDataType nullableArg0 = agg.getCluster().getTypeFactory().createTypeWithNullability(arg0Type, true);
                 return AggregateCall.create(

@@ -109,14 +109,24 @@ aggregate_functions:
   - name: approx_percentile_cont
     description: >-
       PPL `percentile_approx(field, percentile)` — t-digest based approximate
-      percentile. Maps to DataFusion's builtin `approx_percentile_cont`.
+      percentile. Maps to DataFusion's builtin `approx_percentile_cont`. The
+      three-arg overload carries an explicit centroid count (DSL percentiles
+      `tdigest.compression`, PERCENTILE_APPROX_N).
     impls:
       - args:
           - name: x
             value: any1
           - name: percentile
             value: any2
         return: any1
+      - args:
+          - name: x
+            value: any1
+          - name: percentile
+            value: any2
+          - name: centroids
+            value: any3
+        return: any1
   - name: os_count_distinct
     description: >-
       Exact distinct count for use in a window context. PPL

@@ -44,17 +44,17 @@
  * in the coordinator fragment; the unpinned lower copy narrows the scan. {@code argList} is untouched.
  *
  * <p>Runs after marking, before CBO — {@code PROJECT_MERGE} (pre-marking) can't re-fuse the copies, and the
- * ER lands between them. Scope = {@code PERCENTILE_APPROX} and {@code TAKE}, the only aggregates that
- * materialize a literal config arg as a Project column (FIRST/LAST drop their optional N; others carry no
- * literal). Fires only when such a call references a literal column and the Project also has a non-literal
- * column to push down.
+ * ER lands between them. Scope = {@code PERCENTILE_APPROX} / {@code PERCENTILE_APPROX_N} and {@code TAKE},
+ * the only aggregates that materialize literal config args as Project columns (FIRST/LAST drop their
+ * optional N; others carry no literal). Fires only when such a call references a literal column and the
+ * Project also has a non-literal column to push down.
  *
  * @opensearch.internal
  */
 public class OpenSearchAggLiteralArgProjectSplitRule extends RelOptRule {
 
-    // Aggregates whose literal config arg the DataFusion converter re-inlines from the attached Project.
-    private static final java.util.Set<String> LITERAL_ARG_AGGS = java.util.Set.of("PERCENTILE_APPROX", "TAKE");
+    // Aggregates whose literal config args the DataFusion converter re-inlines from the attached Project.
+    private static final java.util.Set<String> LITERAL_ARG_AGGS = java.util.Set.of("PERCENTILE_APPROX", "PERCENTILE_APPROX_N", "TAKE");
 
     public OpenSearchAggLiteralArgProjectSplitRule() {
         super(operand(OpenSearchAggregate.class, operand(OpenSearchProject.class, none())), "OpenSearchAggLiteralArgProjectSplitRule");

@@ -78,14 +78,13 @@ public TransportDslExecuteAction(
     @Override
     protected void doExecute(Task task, SearchRequest request, ActionListener<SearchResponse> listener) {
         threadPool.executor(ThreadPool.Names.SEARCH).execute(() -> {
+            final long startTime = System.currentTimeMillis();
             final QueryPlans plans;
-            final long convertTime;
+            final SearchSourceConverter converter;
             try {
                 String indexName = resolveToSingleIndex(request);
-                long convertStart = System.nanoTime();
-                SearchSourceConverter converter = new SearchSourceConverter(contextProvider.getContext().schema());
+                converter = new SearchSourceConverter(contextProvider.getContext().schema());
                 plans = converter.convert(request.source(), indexName);
-                convertTime = System.nanoTime() - convertStart;
             } catch (Exception e) {
                 logger.error("DSL conversion failed", e);
                 listener.onFailure(e);
@@ -94,7 +93,8 @@ protected void doExecute(Task task, SearchRequest request, ActionListener<Search
             planExecutor.execute(plans, ActionListener.wrap(results -> {
                 final SearchResponse response;
                 try {
-                    response = SearchResponseBuilder.build(results, convertTime);
+                    long tookInMillis = System.currentTimeMillis() - startTime;
+                    response = SearchResponseBuilder.build(results, request, converter.getAggregationRegistry(), tookInMillis);
                 } catch (Exception buildEx) {
                     logger.error("DSL response building failed", buildEx);
                     listener.onFailure(buildEx);

@@ -31,6 +31,7 @@ public class AggregationMetadata {
     private final List<AggregateCall> aggregateCalls;
     private final List<String> aggregateFieldNames;
     private final List<BucketOrder> bucketOrders;
+    private final List<LiteralColumn> literalColumns;
 
     /**
      * Creates aggregation metadata.
@@ -40,19 +41,23 @@ public class AggregationMetadata {
      * @param aggregateCalls Calcite aggregate calls (AVG, SUM, etc.)
      * @param aggregateFieldNames output names for aggregate results
      * @param bucketOrders bucket orders for post-aggregation sorting
+     * @param literalColumns literal-derived columns to append to the aggregate's input, in
+     *                       allocation order (see {@link LiteralColumnAllocator})
      */
     public AggregationMetadata(
         ImmutableBitSet groupByBitSet,
         List<String> groupByFieldNames,
         List<AggregateCall> aggregateCalls,
         List<String> aggregateFieldNames,
-        List<BucketOrder> bucketOrders
+        List<BucketOrder> bucketOrders,
+        List<LiteralColumn> literalColumns
     ) {
         this.groupByBitSet = groupByBitSet;
         this.groupByFieldNames = List.copyOf(groupByFieldNames);
         this.aggregateCalls = List.copyOf(aggregateCalls);
         this.aggregateFieldNames = List.copyOf(aggregateFieldNames);
         this.bucketOrders = List.copyOf(bucketOrders);
+        this.literalColumns = List.copyOf(literalColumns);
     }
 
     /** Returns the GROUP BY column indices. */
@@ -84,4 +89,9 @@ public List<BucketOrder> getBucketOrders() {
     public boolean hasBucketOrders() {
         return !bucketOrders.isEmpty();
     }
+
+    /** Returns the literal-derived columns to append to the aggregate's input, in allocation order. */
+    public List<LiteralColumn> getLiteralColumns() {
+        return literalColumns;
+    }
 }
@@ -13,6 +13,7 @@
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
 import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.calcite.util.ImmutableBitSet;
@@ -37,6 +38,7 @@ public class AggregationMetadataBuilder {
     private final List<AggregateCall> aggregateCalls = new ArrayList<>();
     private final List<String> aggregateFieldNames = new ArrayList<>();
     private final List<BucketOrder> bucketOrders = new ArrayList<>();
+    private final List<LiteralColumn> literalColumns = new ArrayList<>();
     private boolean implicitCountRequested = false;
 
     /** Creates a new empty builder. */
@@ -85,6 +87,41 @@ public void requestImplicitCount() {
         this.implicitCountRequested = true;
     }
 
+    /**
+     * Returns a literal-column allocator. Allocated columns are appended after the
+     * {@code baseFieldCount} input fields (deduplicated) and materialized by the converter
+     * in a pre-aggregate project.
+     *
+     * @param baseFieldCount the field count of the aggregate's un-projected input
+     */
+    public LiteralColumnAllocator literalColumnAllocator(int baseFieldCount) {
+        return new LiteralColumnAllocator() {
+            @Override
+            public int columnFor(double value) {
+                return indexFor(LiteralColumn.constant(value));
+            }
+
+            @Override
+            public int integerColumnFor(long value) {
+                return indexFor(LiteralColumn.integerConstant(value));
+            }
+
+            @Override
+            public int coalescedColumnFor(int fieldIndex, double missingValue) {
+                return indexFor(LiteralColumn.coalesced(fieldIndex, missingValue));
+            }
+
+            private int indexFor(LiteralColumn column) {
+                int existing = literalColumns.indexOf(column);
+                if (existing >= 0) {
+                    return baseFieldCount + existing;
+                }
+                literalColumns.add(column);
+                return baseFieldCount + literalColumns.size() - 1;
+            }
+        };
+    }
+
     /** Returns true if this builder has at least one aggregate call or implicit count. */
     public boolean hasAggregateCalls() {
         return !aggregateCalls.isEmpty() || implicitCountRequested;
@@ -113,8 +150,24 @@ public AggregationMetadata build(RelDataType inputRowType, RelDataTypeFactory ty
         boolean noGroupBy = groupings.isEmpty();
         List<AggregateCall> allCalls = new ArrayList<>();
         for (AggregateCall call : aggregateCalls) {
-            if (noGroupBy) {
-                RelDataType nullableType = typeFactory.createTypeWithNullability(call.getType(), true);
+            boolean isCount = call.getAggregation().getKind() == SqlKind.COUNT;
+            String functionName = call.getAggregation().getName();
+            // Percentile results are nullable even under GROUP BY (engine may emit NULL per group).
+            boolean forcedNullable = "PERCENTILE_APPROX".equals(functionName) || "PERCENTILE_APPROX_N".equals(functionName);
+            RelDataType targetType;
+            if (isCount) {
+                // COUNT is always BIGINT NOT NULL regardless of the input field's type; normalize here
+                // (translators have no type factory) — LogicalAggregate.create asserts the type matches.
+                targetType = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.BIGINT), false);
+            } else if (noGroupBy || forcedNullable) {
+                // AVG, MIN, MAX, SUM return null for empty sets when there's no GROUP BY.
+                targetType = typeFactory.createTypeWithNullability(call.getType(), true);
+            } else {
+                targetType = call.getType();
+            }
+            if (targetType.equals(call.getType())) {
+                allCalls.add(call);
+            } else {
                 allCalls.add(
                     AggregateCall.create(
                         call.getAggregation(),
@@ -124,12 +177,10 @@ public AggregationMetadata build(RelDataType inputRowType, RelDataTypeFactory ty
                         call.getArgList(),
                         call.filterArg,
                         call.getCollation(),
-                        nullableType,
+                        targetType,
                         call.getName()
                     )
                 );
-            } else {
-                allCalls.add(call);
             }
         }
         List<String> allFieldNames = new ArrayList<>(aggregateFieldNames);
@@ -144,14 +195,21 @@ public AggregationMetadata build(RelDataType inputRowType, RelDataTypeFactory ty
                     List.of(),
                     -1,
                     RelCollations.EMPTY,
-                    typeFactory.createSqlType(SqlTypeName.BIGINT),
+                    typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.BIGINT), false),
                     IMPLICIT_COUNT_NAME
                 )
             );
             allFieldNames.add(IMPLICIT_COUNT_NAME);
         }
 
-        return new AggregationMetadata(ImmutableBitSet.of(allGroupIndices), allGroupFieldNames, allCalls, allFieldNames, bucketOrders);
+        return new AggregationMetadata(
+            ImmutableBitSet.of(allGroupIndices),
+            allGroupFieldNames,
+            allCalls,
+            allFieldNames,
+            bucketOrders,
+            literalColumns
+        );
     }
 
     /**

@@ -10,9 +10,13 @@
 
 import org.opensearch.dsl.aggregation.bucket.TermsBucketTranslator;
 import org.opensearch.dsl.aggregation.metric.AvgMetricTranslator;
+import org.opensearch.dsl.aggregation.metric.ExtendedStatsMetricTranslator;
 import org.opensearch.dsl.aggregation.metric.MaxMetricTranslator;
 import org.opensearch.dsl.aggregation.metric.MinMetricTranslator;
+import org.opensearch.dsl.aggregation.metric.PercentilesMetricTranslator;
+import org.opensearch.dsl.aggregation.metric.StatsMetricTranslator;
 import org.opensearch.dsl.aggregation.metric.SumMetricTranslator;
+import org.opensearch.dsl.aggregation.metric.ValueCountMetricTranslator;
 
 /**
  * Creates an {@link AggregationRegistry} populated with all supported translators.
@@ -28,8 +32,11 @@ public static AggregationRegistry create() {
         registry.register(new SumMetricTranslator());
         registry.register(new MinMetricTranslator());
         registry.register(new MaxMetricTranslator());
+        registry.register(new StatsMetricTranslator());
+        registry.register(new ExtendedStatsMetricTranslator());
+        registry.register(new ValueCountMetricTranslator());
+        registry.register(new PercentilesMetricTranslator());
         registry.register(new TermsBucketTranslator());
-        // TODO: add other aggregation translators
         return registry;
     }
 }