Dsl pipeline aggs

[sachin-27]

Description

Adds avg_bucket sibling pipeline aggregation support, computed engine-side. Supersedes #21201, following its plan-composition strategy.

Each sibling pipeline becomes a second plan wrapping the sibling's aggregate:

LogicalAggregate(group=[{}], avg_brand_sales=[AVG($1)])      -- one call per pipeline
  LogicalProject(..., total_sales=[CAST($1):DOUBLE], ...)    -- gap policy + widening
    LogicalSort(sort0=[$2], dir0=[DESC], fetch=[10])         -- sibling's own order + size
      <sibling aggregate>

The filter/sort/fetch shaping mirrors the sibling's visible buckets — vanilla runs sibling pipelines post-truncation, so the pipeline must aggregate what the response shows, not every group. Plans are tagged QueryPlans.Type.PIPELINE; the single result row maps back to pipelines by column name and renders via vanilla's InternalSimpleValue (empty sibling → "value": null).

Key semantics:

• gap_policy: skip = SQL AVG's native NULL handling (excluded from numerator and denominator); insert_zeros = COALESCE(metric, 0).
• Metric columns widen to DOUBLE so the engine's AVG decomposition divides in floating point.
• Pipelines targeting one sibling share one plan — one extra sibling computation regardless of pipeline count.
• Behavior change: previously pipelines were silently dropped; unsupported types, bad paths, and nested pipelines now fail at conversion with descriptive errors.

Scope: avg_bucket over root-level terms siblings, single-level buckets_path (metric or _count), both gap policies. Everything else is rejected explicitly; other *_bucket types are follow-ups reusing this machinery.

Testing

• 14 unit tests (path validation, rejections, same-sibling merge, result conversion).
• 6 golden files, including truncation parity and same-field siblings with different orders; golden infra extended with optional additionalPlans for multi-plan scenarios.
• Verified end to end through Substrait/DataFusion on a live cluster: exact values for happy path, truncation, both gap policies, _count, empty sibling, and rejection.

Related Issues

Supersedes #21201

---

[github-actions]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

📝 TODO sections sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/result/AggregationResponseBuilder.java [77]: Support pipeline aggregations post-processing; currently ignored. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/result/AggregationResponseBuilder.java [252]: Avoid re-scanning full row list on every recursion; index rows by parent bucket key. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/result/SearchResponseBuilder.java [62]: Shard counts, timed_out, and engine-side took require execution metadata. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/result/SearchResponseBuilder.java [69]: Build hits from HITS results. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/metric/MetricTranslator.java [44]: Revisit signature — accept a stream/iterator for bulk conversion, use Arrow-native types.

🔀 Multiple PR themes Sub-PR theme: Implement toInternalAggregation for metric and bucket translators Relevant files: sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/bucket/TermsBucketTranslator.java sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/metric/AbstractMetricTranslator.java sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/metric/AvgMetricTranslator.java sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/metric/MaxMetricTranslator.java sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/metric/MinMetricTranslator.java sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/metric/SumMetricTranslator.java sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/metric/MetricTranslator.java sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/AggregationTranslator.java Sub-PR theme: Add avg_bucket sibling pipeline aggregation support Relevant files: sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/pipeline/AvgBucketTranslator.java sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/pipeline/BucketsPathResolver.java sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/pipeline/MetricColumnPreparer.java sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/pipeline/PipelinePlanComposer.java sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/pipeline/PipelineRegistry.java sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/pipeline/PipelineTranslator.java sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/pipeline/package-info.java

⚡ Recommended focus areas for review Uninitialized variable on failure path converter is declared final and assigned inside the try block after resolveToSingleIndex(request). If resolveToSingleIndex throws, converter is never assigned, but the compiler may still allow the code because control returns from the catch block. However, if the flow is refactored, or if any exception is thrown after converter assignment but before plans assignment, the outer lambda captures the same final variable used later in SearchResponseBuilder.build. Also, startNanos is captured for tookInMillis, but on the failure path the timing is discarded — verify this is intentional (previously convertTime covered only conversion). final long startNanos = System.nanoTime(); final QueryPlans plans; final SearchSourceConverter converter; try { String indexName = resolveToSingleIndex(request); converter = new SearchSourceConverter(contextProvider.getContext().schema()); plans = converter.convert(request.source(), indexName); } catch (Exception e) { logger.error("DSL conversion failed", e); listener.onFailure(e); return; } planExecutor.execute(plans, ActionListener.wrap(results -> { final SearchResponse response; try { long tookInMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos); response = SearchResponseBuilder.build( results, request, converter.getAggregationRegistry(), converter.getPipelineRegistry(), tookInMillis ); Possible NPE on sibling.getType() findSibling returns non-null (throws otherwise), but the instanceof check (sibling instanceof TermsAggregationBuilder) == false then calls sibling.getType() in the error message. This is fine for concrete builders, but if a future user aggregation returns null from getType() the NPE would mask the actual validation error. Minor. AggregationBuilder sibling = findSibling(pipeline, rootAggs, siblingElement.name); if ((sibling instanceof TermsAggregationBuilder) == false) { throw new ConversionException( "pipeline aggregation [" + pipeline.getName() + "] sibling [" + sibling.getName() + "] of type [" + sibling.getType() + "] is not supported; only [terms] siblings are supported" ); } Missing null check for pipeline builder lookup findPipelineBuilder iterates request.source().aggregations().getPipelineAggregatorFactories(). If a PIPELINE ExecutionResult is present but the request has no pipelines (or aggregations() is null), appendPipelineResults filters early, but findPipelineBuilder itself dereferences request.source().aggregations() unconditionally. This is safe today due to the earlier early-return, but is a fragile coupling — any refactor that removes the early-return will NPE. private static PipelineAggregationBuilder findPipelineBuilder(String name, SearchRequest request) throws ConversionException { for (PipelineAggregationBuilder pipeline : request.source().aggregations().getPipelineAggregatorFactories()) { if (pipeline.getName().equals(name)) { return pipeline; } } throw new ConversionException("Pipeline result column [" + name + "] has no matching pipeline aggregation in the request"); } Doc count error not computed StringTerms is constructed with docCountError=0 and showTermDocCountError=false unconditionally. For distributed executions where terms are shard-approximate, this hides potential errors. If the analytics engine guarantees exact counts (no shard fan-out) this is correct; otherwise doc_count_error_upper_bound in the response will be misleading. Confirm this assumption matches the execution model. ); return new StringTerms( agg.getName(), order, order, AggregationTranslator.userMetadata(agg), DocValueFormat.RAW, agg.shardSize(), false, otherDocCount, termBuckets, 0, thresholds );

[github-actions]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Key sibling map by unique name Using TermsAggregationBuilder as a HashMap/LinkedHashMap key relies on its equals/hashCode semantics. If two distinct sibling terms builders in the request happen to be equal (same field, same options, same name), they would collide into one entry, merging pipelines that target different siblings. Key by aggregation name (which is unique among root aggs) instead to avoid equality-based collisions. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/converter/SearchSourceConverter.java [186-197] -Map<TermsAggregationBuilder, List<PipelinePlanComposer.PipelineTarget>> bySibling = new LinkedHashMap<>(); +Map<String, TermsAggregationBuilder> siblingByName = new LinkedHashMap<>(); +Map<String, List<PipelinePlanComposer.PipelineTarget>> bySiblingName = new LinkedHashMap<>(); for (PipelineAggregationBuilder pipeline : pipelines) { PipelineTranslator<PipelineAggregationBuilder> translator = pipelineRegistry.get(pipeline.getClass()); if (translator == null) { throw new ConversionException( "pipeline aggregation [" + pipeline.getName() + "] of type [" + pipeline.getWriteableName() + "] is not supported" ); } BucketsPathResolver.ResolvedBucketsPath resolved = BucketsPathResolver.resolve(pipeline, rootAggs, aggRegistry); - bySibling.computeIfAbsent(resolved.sibling(), s -> new ArrayList<>()) + siblingByName.putIfAbsent(resolved.sibling().getName(), resolved.sibling()); + bySiblingName.computeIfAbsent(resolved.sibling().getName(), s -> new ArrayList<>()) .add(new PipelinePlanComposer.PipelineTarget(pipeline, resolved.metricColumn())); } Suggestion importance[1-10]: 5 __ Why: Valid concern about relying on TermsAggregationBuilder.equals/hashCode for map keys; keying by unique aggregation name is more robust. However, in practice OpenSearch enforces unique aggregation names at the root level, making collisions unlikely.	Low
General	Guard bucket truncation against invalid size agg.size() returns an int, but if the user configures a very large size or if size is negative/zero due to misconfiguration, subList(0, agg.size()) could throw IndexOutOfBoundsException when agg.size() exceeds termBuckets.size() in an edge case, or truncate incorrectly. Also, when agg.size() <= 0, this bypasses the truncation branch entirely and returns all buckets. Consider guarding with Math.min(agg.size(), termBuckets.size()) and validating size is positive. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/bucket/TermsBucketTranslator.java [83-89] long otherDocCount = 0; -if (termBuckets.size() > agg.size()) { - for (int i = agg.size(); i < termBuckets.size(); i++) { +int effectiveSize = Math.max(0, agg.size()); +if (termBuckets.size() > effectiveSize) { + for (int i = effectiveSize; i < termBuckets.size(); i++) { otherDocCount += termBuckets.get(i).getDocCount(); } - termBuckets = new ArrayList<>(termBuckets.subList(0, agg.size())); + termBuckets = new ArrayList<>(termBuckets.subList(0, effectiveSize)); } Suggestion importance[1-10]: 3 __ Why: agg.size() in TermsAggregationBuilder is validated to be positive at build time, so this defensive guard addresses an unlikely edge case. The subList call is safe since it's guarded by the size comparison.	Low
	Guard against short pipeline result rows When row is non-null but has fewer columns than fieldNames, row[i] would throw ArrayIndexOutOfBoundsException. Add a bounds check to defensively handle mismatched row width, or assert the invariant that row length equals field count. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/result/SearchResponseBuilder.java [132-143] for (int i = 0; i < fieldNames.size(); i++) { PipelineAggregationBuilder pipeline = findPipelineBuilder(fieldNames.get(i), request); PipelineTranslator<PipelineAggregationBuilder> translator = pipelineRegistry.get(pipeline.getClass()); if (translator == null) { throw new ConversionException( "No pipeline translator registered for [" + pipeline.getName() + "] of type [" + pipeline.getWriteableName() + "]" ); } - Object cell = row == null ? null : row[i]; + Object cell = (row == null || i >= row.length) ? null : row[i]; combined.add(translator.toInternalAggregation(pipeline, cell)); } Suggestion importance[1-10]: 3 __ Why: A defensive bounds check for a scenario that should not occur given the invariant that row length matches field count. Low impact but slightly improves robustness.	Low

[github-actions]

❌ Gradle check result for 32a720f: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?