Build aggregation responses from execution results

[sachin-27]

Convert flat per-granularity execution results into the client's nested aggregation response using the original request as template.

Description

Converts analytics engine execution results into OpenSearch InternalAggregations format, enabling aggregation responses in the DSL query executor. Supersedes #21346.

Changes

• AggregationResponseBuilder: Converts flat execution results to nested aggregation structure using granularity-based matching
• ComparisonUtils: Type coercion for numeric comparisons
• Metric/Terms translators: Implement response conversion (InternalAvg/Sum/Min/Max, StringTerms)
• TransportDslExecuteAction / SearchResponseBuilder: Wire the response builder into the search flow

Verified end to end on a live composite index: terms + metrics, filtered global metrics, and two-level nested aggregations return correct legacy-shaped responses.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 518338d)

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/SearchResponseBuilder.java [54]: shard counts, timed_out, and took require execution metadata from analytics plugin sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/result/SearchResponseBuilder.java [61]: Build hits from HITS results

🔀 No multiple PR themes

⚡ Recommended focus areas for review Granularity Key Collision granularityKey sorts group field names and joins them with a NUL separator. Two different groupings whose field names differ only in ordering after sort (e.g., grouping by ["a","b"] vs grouping by ["a","b"] in different nesting orders) intentionally collide — but so would ["ab","c"] vs ["a","bc"] if a field name happens to contain NUL. More importantly, if the same field name appears at multiple nesting levels (which is legal in nested aggregations that re-group by the same field with different filters), sorting collapses them to a single key and the wrong ExecutionResult may be selected. Consider validating that grouping paths are uniquely resolvable. * Canonical granularity key: sorted, comma-joined group field names. Sorted because the * plan emits group columns in schema order while the request walk accumulates them in * nesting order — the two must produce the same key. */ private static String granularityKey(List<String> groupFieldNames) { return groupFieldNames.stream().sorted().collect(Collectors.joining(AGGREGATION_LEVEL_SEPARATOR)); } Possible Issue buildBucket reads _count from firstRowInGroup but with nested sub-aggregations the same bucket key can appear on multiple rows (one per sub-group) with different _count values that need to be aggregated. Using only the first row's _count will under-report doc_count for parent buckets when the granularity used has finer grouping than the current bucket level. Verify whether the finer-granularity result is actually being queried here (it appears to be, since newGroupFields extends with sub-grouping) — if so, doc_count should be summed across grouped rows rather than taken from the first row. Integer countIdx = colIndex.get("_count"); if (countIdx == null) { throw new ConversionException("Missing _count column in aggregation result"); } Object[] firstRowInGroup = entry.getValue().get(0); long docCount = ((Number) firstRowInGroup[countIdx]).longValue(); InternalAggregations subAggregations = subAggs.isEmpty() ? InternalAggregations.EMPTY : InternalAggregations.from(buildLevel(subAggs, newGroupFields, childFilter)); buckets.add(new BucketEntry(entry.getKey(), docCount, subAggregations)); } Semantic Concern Encoding the average as (sum=value, count=1) reproduces getValue() correctly for direct use, but any downstream logic that inspects getSum() or getCount() (e.g., cross-shard reduce, pipeline aggregations like avg_bucket, or serialization consumers) will see fabricated values. If these InternalAvg instances ever flow through reduction or pipeline agg paths, results will be wrong. Confirm the analytics response is terminal and not further reduced. @Override public InternalAggregation toInternalAggregation(String name, Object value) { if (value == null) { return new InternalAvg(name, 0.0, 0, DocValueFormat.RAW, null); } return new InternalAvg(name, toDouble(value), 1, DocValueFormat.RAW, null); }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 518338d
Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Guard against non-string terms keys Coercing every key via toString() will produce incorrect string representations for non-string types (e.g. numeric terms aggregations render Long/Double values into a StringTerms instead of the appropriate LongTerms/DoubleTerms), and will conflict with the TermsAggregationBuilder's configured value type. Consider dispatching on key type to build LongTerms/DoubleTerms where applicable, or explicitly validating that the field is string-typed here. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/bucket/TermsBucketTranslator.java [65-74] for (BucketEntry entry : buckets) { Object key = entry.keys().get(0); if (key == null) { - // SQL GROUP BY emits a NULL group; legacy terms excludes docs with a - // missing field entirely (no bucket) unless "missing" is configured. continue; + } + if (!(key instanceof CharSequence)) { + throw new IllegalStateException( + "TermsBucketTranslator currently supports only string keys; got " + key.getClass().getSimpleName() + ); } BytesRef term = new BytesRef(key.toString()); termBuckets.add(new StringTerms.Bucket(term, entry.docCount(), entry.subAggs(), false, 0, DocValueFormat.RAW)); } Suggestion importance[1-10]: 6 __ Why: The class Javadoc already acknowledges that per-key-type Long/Double variants are follow-up work, so failing loudly on non-string keys is a reasonable defensive guard to avoid silently producing incorrect output.	Low
	Fail fast on missing aggregate node When the plan has no LogicalAggregate (e.g. a metric-only aggregation without GROUP BY), this method returns NO_GROUPING_KEY, which collides with the key used for both no-grouping metrics and any aggregate that produced an empty group list. However, if multiple ExecutionResults produce the same key (e.g. a top-level metric plan and a plan whose aggregate could not be located), the last one silently overwrites earlier ones in granularityMap. Consider detecting a collision in the constructor and failing fast, or preserving all results per key, to prevent silent result loss. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/result/AggregationResponseBuilder.java [285-297] private static String computeGranularityKey(ExecutionResult result) { RelNode node = result.getPlan().relNode(); while (node != null && !(node instanceof LogicalAggregate)) { node = node.getInputs().isEmpty() ? null : node.getInput(0); } if (node instanceof LogicalAggregate agg) { - // Resolve group field names against the aggregate's INPUT row type via the group - // set — robust against output-side renames, unlike the aggregate's own row type. List<String> inputFields = agg.getInput().getRowType().getFieldNames(); return granularityKey(agg.getGroupSet().asList().stream().map(inputFields::get).collect(Collectors.toList())); } - return NO_GROUPING_KEY; + throw new IllegalStateException("Aggregation ExecutionResult has no LogicalAggregate in plan"); } Suggestion importance[1-10]: 4 __ Why: Reasonable defensive change, but NO_GROUPING_KEY is also used intentionally for the metric-only no-grouping case, so throwing might be too strict. The suggestion notes a plausible edge case but its impact is limited.	Low
General	Detect duplicate granularity keys granularityMap.put silently overwrites when two ExecutionResults have the same granularity key (which can happen if two subtrees produce the same sorted set of group fields, e.g. two sibling terms aggs on the same field). Detect duplicates and reject them explicitly instead of dropping one silently, otherwise nested/sibling aggregations may return incorrect buckets. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/result/AggregationResponseBuilder.java [45-52] public AggregationResponseBuilder(AggregationRegistry registry, List<ExecutionResult> aggResults) { this.registry = registry; this.granularityMap = new HashMap<>(); for (ExecutionResult result : aggResults) { String key = computeGranularityKey(result); - granularityMap.put(key, result); + if (granularityMap.putIfAbsent(key, result) != null) { + throw new IllegalStateException("Duplicate granularity key for aggregation results: [" + key + "]"); + } } } Suggestion importance[1-10]: 6 __ Why: Valid concern — sibling aggregations sharing the same group-field set would collide and silently overwrite. Failing fast improves correctness detection, though it may not fully solve the underlying design issue.	Low
	Avg encoding may break reduction Encoding the final average as (sum=value, count=1) breaks aggregation reduction: if InternalAvg.reduce is ever invoked (e.g. cross-cluster or pipeline aggregations that consume InternalAvg), it will sum values and counts across shards and produce an incorrect average. Consider documenting/enforcing that these results are terminal and never reduced, or emitting a subclass/format that guards against reduction. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/metric/AvgMetricTranslator.java [45-50] @Override public InternalAggregation toInternalAggregation(String name, Object value) { + // NOTE: terminal-only encoding — must not be passed to InternalAvg.reduce(). if (value == null) { return new InternalAvg(name, 0.0, 0, DocValueFormat.RAW, null); } return new InternalAvg(name, toDouble(value), 1, DocValueFormat.RAW, null); } Suggestion importance[1-10]: 5 __ Why: Legitimate concern about reduction semantics, but the improved code only adds a comment. The suggestion doesn't materially fix the issue.	Low

---

Previous suggestions

Suggestions up to commit c6a65de

Category	Suggestion	Impact
Possible issue	Use input field names for grouping key Using getRowType().getFieldNames() from the LogicalAggregate may return output field names (renamed via aliases), while the request-side walk accumulates original schema field names. This can cause granularity key mismatches. Consider retrieving group field names from the aggregate's input row type at the grouped column indices instead. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/result/AggregationResponseBuilder.java [284-294] private static String computeGranularityKey(ExecutionResult result) { RelNode node = result.getPlan().relNode(); while (node != null && !(node instanceof LogicalAggregate)) { node = node.getInputs().isEmpty() ? null : node.getInput(0); } if (node instanceof LogicalAggregate agg) { - int groupCount = agg.getGroupCount(); - return granularityKey(agg.getRowType().getFieldNames().stream().limit(groupCount).collect(Collectors.toList())); + List<String> inputFields = agg.getInput().getRowType().getFieldNames(); + List<String> groupFields = agg.getGroupSet().asList().stream() + .map(inputFields::get) + .collect(Collectors.toList()); + return granularityKey(groupFields); } return NO_GROUPING_KEY; } Suggestion importance[1-10]: 6 __ Why: Potentially valid concern about alias/rename mismatches between aggregate output field names and the request-side accumulated field names, which could affect granularity key matching. Correctness depends on plan structure not shown.	Low
General	Preserve distinction between null and empty keys Rendering null keys as empty string "" is indistinguishable from a bucket with an actual empty-string key, which can cause data collisions when both null and empty values exist. Consider filtering out null keys, or use a distinct representation (e.g., using DocValueFormat's null handling) to preserve the distinction, matching OpenSearch's native missing-value bucket semantics. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/bucket/TermsBucketTranslator.java [64-69] List<StringTerms.Bucket> termBuckets = new ArrayList<>(); for (BucketEntry entry : buckets) { Object key = entry.keys().get(0); - BytesRef term = new BytesRef(key == null ? "" : key.toString()); + if (key == null) { + continue; // null keys are excluded unless a "missing" value is configured + } + BytesRef term = new BytesRef(key.toString()); termBuckets.add(new StringTerms.Bucket(term, entry.docCount(), entry.subAggs(), false, 0, DocValueFormat.RAW)); } Suggestion importance[1-10]: 5 __ Why: Valid observation that mapping null to empty string may cause key collisions with actual empty strings, though the correct handling depends on OpenSearch's missing-value semantics for terms aggregations.	Low
	Guard against null _count values The _count column is looked up inside the group loop but is invariant per result — hoist it above the loop to fail fast and avoid repeated map lookups. Additionally, if the value at countIdx is null (possible with outer joins or empty groups), the cast to Number will throw NPE with an unhelpful message. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/result/AggregationResponseBuilder.java [182-187] Integer countIdx = colIndex.get("_count"); if (countIdx == null) { throw new ConversionException("Missing _count column in aggregation result"); } +... Object[] firstRowInGroup = entry.getValue().get(0); -long docCount = ((Number) firstRowInGroup[countIdx]).longValue(); +Object countValue = firstRowInGroup[countIdx]; +long docCount = countValue == null ? 0L : ((Number) countValue).longValue(); Suggestion importance[1-10]: 4 __ Why: Defensive null-check for _count value; unlikely in aggregation results but a minor robustness improvement. The hoisting claim is inaccurate since the lookup is already outside the inner logic.	Low
	Avoid NPE in error message construction Calling value.getClass() when value is null will throw a NullPointerException masking the actual issue. Since callers already handle null before calling this method, but a defensive null check in the error message would prevent confusing NPE stack traces if null slips through. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/metric/AbstractMetricTranslator.java [75-80] protected static double toDouble(Object value) { if (value instanceof Number number) { return number.doubleValue(); } - throw new IllegalStateException("Expected numeric aggregation result but got " + value.getClass().getSimpleName()); + throw new IllegalStateException( + "Expected numeric aggregation result but got " + (value == null ? "null" : value.getClass().getSimpleName()) + ); } Suggestion importance[1-10]: 3 __ Why: Minor defensive improvement; callers guard against null before invoking toDouble, so the NPE risk is low. Still, a small readability/robustness win.	Low

[github-actions]

❌ Gradle check result for c6a65de: 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?

[github-actions]

Persistent review updated to latest commit 518338d

[github-actions]

❌ Gradle check result for 518338d: 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?