Add stats, extended_stats, and value_count support to DSL Calcite plugin

[sachin-27]

Description

Adds support for stats, extended_stats, and value_count metric aggregations to the DSL Calcite plugin.
Stacks on #22526 — only the last commit is new to this PR

Key changes:

• Refactored MetricTranslator interface to support multi-metric aggregations (returns List instead of single call)
• Implemented StatsMetricTranslator (count, min, max, sum)
• Implemented ExtendedStatsMetricTranslator (count, min, max, sum, variance)
• Implemented ValueCountMetricTranslator (count on specific field)
• Made COUNT type explicitly non-nullable to match SQL semantics

Note:

Derived metrics like avg and std_deviation are computed by InternalStats/InternalExtendedStats from the base metrics rather than in SQL.

Tests:

Added unit tests for all three new translators covering valid values, null handling, and error cases.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit b7abe1f)

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/aggregation/metric/MetricTranslator.java [142]: Revisit signature — accept a stream/iterator for bulk conversion; use Arrow-native types once available. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/result/SearchResponseBuilder.java [54]: Shard counts, timed_out, and engine-side 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.

🔀 Multiple PR themes Sub-PR theme: Implement TermsBucketTranslator.toBucketAggregation Relevant files: sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/bucket/TermsBucketTranslator.java sandbox/plugins/dsl-query-executor/src/test/java/org/opensearch/dsl/aggregation/bucket/TermsBucketTranslatorTests.java Sub-PR theme: Aggregation response building infrastructure Relevant files: sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/result/AggregationResponseBuilder.java sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/result/SearchResponseBuilder.java sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/util/ComparisonUtils.java sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/action/TransportDslExecuteAction.java sandbox/plugins/dsl-query-executor/src/test/java/org/opensearch/dsl/result/AggregationResponseBuilderTests.java sandbox/plugins/dsl-query-executor/src/test/java/org/opensearch/dsl/util/ComparisonUtilsTests.java Sub-PR theme: Add stats, extended_stats, and value_count metric translators Relevant files: sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/metric/StatsMetricTranslator.java sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/metric/ExtendedStatsMetricTranslator.java sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/metric/ValueCountMetricTranslator.java sandbox/plugins/dsl-query-executor/src/test/java/org/opensearch/dsl/aggregation/metric/StatsMetricTranslatorTests.java sandbox/plugins/dsl-query-executor/src/test/java/org/opensearch/dsl/aggregation/metric/ExtendedStatsMetricTranslatorTests.java sandbox/plugins/dsl-query-executor/src/test/java/org/opensearch/dsl/aggregation/metric/ValueCountMetricTranslatorTests.java

⚡ Recommended focus areas for review Possible NPE keyString(Object key) calls key.toString() without a null check. Although toBucketAggregation filters null keys from the top-level list before dispatching, keyString is called from stringTerms on the already-filtered list, so this is safe for the current path. However, in longTerms/doubleTerms the code casts entry.keys().get(0) to Number/Boolean without null checks either — if a null key ever reaches these branches (e.g., a future caller bypasses the filter), a NullPointerException will occur. Consider defensive null handling or documenting the precondition explicitly. private static InternalAggregation longTerms(TermsAggregationBuilder agg, List<BucketEntry> entries, DocValueFormat format) { List<LongTerms.Bucket> termBuckets = new ArrayList<>(entries.size()); for (BucketEntry entry : entries) { Object key = entry.keys().get(0); long term = key instanceof Boolean bool ? (bool ? 1L : 0L) : ((Number) key).longValue(); termBuckets.add(new LongTerms.Bucket(term, entry.docCount(), entry.subAggs(), false, 0, format)); } BucketOrder order = agg.order(); return new LongTerms(agg.getName(), order, order, null, format, agg.shardSize(), false, 0, termBuckets, 0, thresholds(agg)); } private static InternalAggregation doubleTerms(TermsAggregationBuilder agg, List<BucketEntry> entries) { List<DoubleTerms.Bucket> termBuckets = new ArrayList<>(entries.size()); for (BucketEntry entry : entries) { double term = ((Number) entry.keys().get(0)).doubleValue(); termBuckets.add(new DoubleTerms.Bucket(term, entry.docCount(), entry.subAggs(), false, 0, DocValueFormat.RAW)); } BucketOrder order = agg.order(); return new DoubleTerms( agg.getName(), order, order, null, DocValueFormat.RAW, agg.shardSize(), false, 0, termBuckets, 0, thresholds(agg) ); } /** Binary keys are ip columns: render the address string like classic ip terms. */ private static String keyString(Object key) { if (key instanceof byte[] bytes) { try { return InetAddress.getByAddress(bytes).getHostAddress(); } catch (UnknownHostException e) { // Not a 4/16-byte address; fall back to a printable, deterministic form. return Base64.getEncoder().encodeToString(bytes); } } return key.toString(); } Incorrect variance for empty set When count == 0, toInternalAggregation passes sumOfSquares=0 and sum=0 to InternalExtendedStats. The test testToInternalAggregationWithNull asserts stats.getVariance() is NaN — which works because InternalExtendedStats computes variance internally from count/sum/sumOfSquares. However, when only some values are present (e.g., only count/sum set as in testToInternalAggregationWithPartialNulls) the code silently defaults missing variance to 0, which will produce a misleading sumOfSquares calculation (0 + avg^2) * count and consequently wrong std_deviation/bounds. Consider whether partial-null inputs should be treated as an error rather than silently substituted. public InternalAggregation toInternalAggregation(ExtendedStatsAggregationBuilder agg, Map<String, Object> values) { String name = agg.getName(); double sigma = agg.sigma(); long count = longValue(values, name + COUNT_SUFFIX); double min = doubleValue(values, name + MIN_SUFFIX, Double.POSITIVE_INFINITY); double max = doubleValue(values, name + MAX_SUFFIX, Double.NEGATIVE_INFINITY); double sum = doubleValue(values, name + SUM_SUFFIX, 0.0); double variance = doubleValue(values, name + VARIANCE_SUFFIX, 0.0); double avg = count > 0 ? sum / count : 0; double sumOfSquares = count > 0 ? (variance + avg * avg) * count : 0; return new InternalExtendedStats(name, count, sum, min, max, sumOfSquares, sigma, MetricTranslator.parseFormat(agg.format()), null); } Literal type mismatch For INTEGER_CONSTANT, makeExactLiteral(BigDecimal.valueOf((long) column.value())) produces a DECIMAL literal, not an integer typed to match the aggregate call's expected argument type. If a translator allocates an integer constant expecting a specific integer type for a COUNT/SUM argument, downstream type validation in Calcite may reject or coerce this unexpectedly. Consider using a typed literal (e.g., makeLiteral(value, typeFactory.createSqlType(SqlTypeName.BIGINT), false)) to ensure the projected column matches the expected type. case INTEGER_CONSTANT -> { projects.add(rexBuilder.makeExactLiteral(BigDecimal.valueOf((long) column.value()))); names.add("_lit" + i); } Wall-clock time for took startTime = System.currentTimeMillis() and later tookInMillis = System.currentTimeMillis() - startTime use wall-clock time, which can go backward if the system clock is adjusted (NTP, DST) during execution, producing negative or incorrect took values. Prefer System.nanoTime() for measuring elapsed durations. final long startTime = System.currentTimeMillis(); 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 = System.currentTimeMillis() - startTime; response = SearchResponseBuilder.build(results, request, converter.getAggregationRegistry(), tookInMillis);

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to b7abe1f

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Preserve exact long value for integer literals LiteralColumn.value() is a double; casting to long silently truncates values outside the exact-integer range of double (|v| > 2^53) or with fractional parts, which corrupts the intended literal. Since LiteralColumn.integerConstant(long) accepts a full long, store the integer value separately (or as BigDecimal) rather than round-tripping through double. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/converter/AggregateConverter.java [68-71] case INTEGER_CONSTANT -> { + // TODO: LiteralColumn should carry the original long to avoid double-precision truncation. projects.add(rexBuilder.makeExactLiteral(BigDecimal.valueOf((long) column.value()))); names.add("_lit" + i); } Suggestion importance[1-10]: 6 __ Why: Valid concern about precision loss when longs exceed 2^53 are stored as double in LiteralColumn. The improved_code only adds a TODO comment without fixing the issue, but the underlying observation is correct and could lead to subtle bugs.	Low
	Fail loudly when aggregate node missing computeGranularityKey only descends the first input, so a plan whose LogicalAggregate is on a non-primary branch (or missing entirely) silently returns NO_GROUPING_KEY, colliding with legitimately-ungrouped results and yielding wrong buckets. Log or throw when no LogicalAggregate is found so misuse fails loudly instead of producing silently incorrect responses. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/result/AggregationResponseBuilder.java [295-307] 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) { + List<String> inputFields = agg.getInput().getRowType().getFieldNames(); + return granularityKey(agg.getGroupSet().asList().stream().map(inputFields::get).collect(Collectors.toList())); + } + throw new IllegalStateException("Execution result plan has no LogicalAggregate: " + result.getPlan().relNode()); +} Suggestion importance[1-10]: 4 __ Why: Reasonable suggestion to fail loudly on unexpected plan structure to aid debugging, but the existing fallback to NO_GROUPING_KEY may be intentional for the no-grouping case. The impact is moderate for debuggability.	Low
General	Compute sum_of_squares directly, not from variance The variance-based derivation of sumOfSquares (SS = (var + avg²) * count) is numerically unstable and can produce noticeably wrong values for large means with small variance due to catastrophic cancellation and floating-point rounding, causing sum_of_squares to diverge from the classic implementation. Consider computing SS directly (e.g., via a dedicated SUM(x*x) aggregate call) rather than back-deriving from population variance. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/metric/ExtendedStatsMetricTranslator.java [101-102] double avg = count > 0 ? sum / count : 0; -double sumOfSquares = count > 0 ? (variance + avg * avg) * count : 0; +// Note: back-deriving SS from VAR_POP is lossy; prefer a dedicated SUM(x*x) aggregate. +double sumOfSquares = count > 0 ? Math.fma(avg, avg * count, variance * count) : 0; Suggestion importance[1-10]: 5 __ Why: Legitimate concern about numerical stability when back-deriving sum_of_squares from variance, which could differ from classic implementation for large means. However, the fix would require engine-level changes (adding a SUM(x*x) aggregate), so this is more a design concern than a directly actionable fix.	Low
	Infer terms type from field, not first key When nonNull is empty, sample is null and the method falls through to stringTerms, which may not match the requested field's declared type (e.g., a numeric field with no matching docs would return StringTerms instead of LongTerms/DoubleTerms). This can cause response-shape inconsistencies for empty results. Consider deriving the terms type from the aggregation builder's value type/field mapping rather than sampling the first key, or at least handle the empty case explicitly. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/bucket/TermsBucketTranslator.java [77-87] -Object sample = nonNull.isEmpty() ? null : nonNull.get(0).keys().get(0); +if (nonNull.isEmpty()) { + // No buckets: return an empty StringTerms (or infer type from the field mapping if available). + return stringTerms(agg, nonNull); +} +Object sample = nonNull.get(0).keys().get(0); if (sample instanceof Boolean) { return longTerms(agg, nonNull, DocValueFormat.BOOLEAN); } if (sample instanceof Double || sample instanceof Float) { return doubleTerms(agg, nonNull); } if (sample instanceof Number) { return longTerms(agg, nonNull, DocValueFormat.RAW); } return stringTerms(agg, nonNull); Suggestion importance[1-10]: 4 __ Why: Valid observation that empty results always fall through to StringTerms regardless of field type, which may cause response-shape inconsistencies. However, the improved_code is essentially identical to the existing behavior (still returns stringTerms for empty) and doesn't actually fix the underlying issue.	Low

---

Previous suggestions

Suggestions up to commit 2233e09

Category	Suggestion	Impact
General	Use constant for implicit count column The _count column name is hardcoded here but defined as AggregationMetadataBuilder.IMPLICIT_COUNT_NAME. Using the constant avoids drift if the name is ever changed. Also add a null-check on the cell value since numeric aggregate results can theoretically be null. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/result/AggregationResponseBuilder.java [193-198] -Integer countIdx = colIndex.get("_count"); +Integer countIdx = colIndex.get(AggregationMetadataBuilder.IMPLICIT_COUNT_NAME); 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]: 5 __ Why: Using the IMPLICIT_COUNT_NAME constant instead of the hardcoded "_count" string improves maintainability and avoids drift. This is a valid code-quality improvement.	Low
	Assert integer constant is integral Casting column.value() (a double) to long silently truncates fractional parts. If an INTEGER_CONSTANT is somehow created with a non-integral double, this hides the bug. Consider storing integer constants as long in LiteralColumn or asserting the value is integral here. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/converter/AggregateConverter.java [68-71] case INTEGER_CONSTANT -> { - projects.add(rexBuilder.makeExactLiteral(BigDecimal.valueOf((long) column.value()))); + double v = column.value(); + assert v == Math.floor(v) && !Double.isInfinite(v) : "INTEGER_CONSTANT must be integral: " + v; + projects.add(rexBuilder.makeExactLiteral(BigDecimal.valueOf((long) v))); names.add("_lit" + i); } Suggestion importance[1-10]: 3 __ Why: LiteralColumn.integerConstant(long) already accepts a long and stores as double, so at the call site the value is integral. The assertion is defensive but low-impact.	Low
	Document floating-point precision limitation Computing sumOfSquares via (variance + avg²) × count accumulates floating-point error, especially for large values where avg² dominates and variance becomes lost precision. This can cause noticeable deviations from the classic path's sum-of-squares value. Consider computing/tracking sum-of-squares more directly if possible, or document that response values may differ slightly from classic results. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/metric/ExtendedStatsMetricTranslator.java [101-102] double avg = count > 0 ? sum / count : 0; +// Note: derived from population variance; may accumulate FP error vs. direct SUM(x^2). double sumOfSquares = count > 0 ? (variance + avg * avg) * count : 0; Suggestion importance[1-10]: 2 __ Why: The suggestion only adds a comment about floating-point precision; the existing code already has a class-level comment explaining the derivation. Marginal improvement.	Low
Possible issue	Guard against mixed-type bucket keys When the sample key is a Boolean but other buckets have non-Boolean keys (or vice-versa), the cast ((Number) key) will throw ClassCastException. Since the type dispatch is only based on the first bucket's key, mixed-type buckets can crash. Consider handling each key defensively or ensuring type consistency across all buckets before dispatching. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/bucket/TermsBucketTranslator.java [116-117] -long term = key instanceof Boolean bool ? (bool ? 1L : 0L) : ((Number) key).longValue(); +long term; +if (key instanceof Boolean bool) { + term = bool ? 1L : 0L; +} else if (key instanceof Number number) { + term = number.longValue(); +} else { + throw new IllegalStateException("Unexpected bucket key type: " + key.getClass().getSimpleName()); +} termBuckets.add(new LongTerms.Bucket(term, entry.docCount(), entry.subAggs(), false, 0, format)); Suggestion importance[1-10]: 3 __ Why: Bucket keys within a single terms aggregation should be type-consistent by construction (schema-typed), so mixed types are unlikely. The suggestion adds defensive error handling but has limited practical impact.	Low

Suggestions up to commit 88231ca

Category	Suggestion	Impact
Possible issue	Reject numeric keys in StringTerms path Using key.toString() for numeric keys (Long/Double) produces string representations like "42" or "3.14" inside a StringTerms, which will not roundtrip correctly through the terms aggregation response format (numeric terms should be LongTerms/DoubleTerms with proper key rendering). At minimum, guard against numeric keys or explicitly reject them until proper per-type bucket support is added, to avoid silently returning wrong-typed results. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/bucket/TermsBucketTranslator.java [72-73] +if (key instanceof Number) { + throw new UnsupportedOperationException("Numeric terms keys not yet supported; only string keys are handled"); +} 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: Valid concern: using key.toString() for numeric bucket keys in StringTerms may produce incorrect response formatting for Long/Double terms. The existing code comment already acknowledges this as follow-up work, so failing fast could prevent silent data corruption, though it may break usage that currently works.	Low
General	Distinguish missing column from null cell HashMap.put rejects null values silently here (actually it doesn't, but callers of values.get cannot distinguish "missing column" from "SQL NULL cell"). More importantly, Map.of is used for expected values in tests, but real execution may yield null cells. Since a SQL NULL cell will be stored via put, this is fine — but downstream StatsMetricTranslator.longValue/doubleValue check values.get(key) == null which conflates "column missing from row" with "cell was NULL". Consider tracking populated keys explicitly or documenting that null-cell and missing-column are treated identically. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/result/AggregationResponseBuilder.java [122-133] Map<String, Object> values = new HashMap<>(); +boolean anyPopulated = false; for (String fieldName : translator.getAggregateFieldNames(agg)) { Integer colIdx = colIndex.get(fieldName); if (colIdx != null) { values.put(fieldName, matchingRow[colIdx]); + anyPopulated = true; } } -if (values.isEmpty()) { +if (!anyPopulated) { return buildEmptyMetric(translator, agg); } return translator.toInternalAggregation(agg, values); Suggestion importance[1-10]: 4 __ Why: The suggestion identifies a subtle semantic conflation between missing columns and null values, but its practical impact is limited since translators handle null cells with sensible empty-set defaults. The improvement is a minor robustness enhancement.	Low
	Document precision loss in derivation Reconstructing sumOfSquares from population variance introduces floating-point error, especially for large counts or values with high magnitude (catastrophic cancellation when variance is small relative to avg^2). This makes getSumOfSquares() and derived variance/std_deviation in the response drift from what a direct SUM(x*x) would give. Consider documenting this precision limitation or requesting SUM_OF_SQUARES support from the engine to compute it directly. sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/metric/ExtendedStatsMetricTranslator.java [92-95] +// NOTE: reconstructing sumOfSquares from VAR_POP loses precision for large counts / +// small variance relative to mean^2 (catastrophic cancellation). Prefer engine-side +// SUM(x*x) when available; downstream variance/std_deviation may drift accordingly. double avg = count > 0 ? sum / count : 0; double sumOfSquares = count > 0 ? (variance + avg * avg) * count : 0; return new InternalExtendedStats(name, count, sum, min, max, sumOfSquares, sigma, DocValueFormat.RAW, null); Suggestion importance[1-10]: 3 __ Why: The suggestion only adds a documentation comment about a known floating-point precision limitation. While the observation is technically accurate, it doesn't change functionality and the existing class-level javadoc already explains the derivation approach.	Low

[github-actions]

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

[aggarwalmayank]

This encoding works only when the response is never reduced across shards or with another InternalAvg.

InternalAvg.reduce sums the sums variable of IntervalAvg and sums the counts variable of IntervalAvg and divides.

two shards each reporting (v, 1) reduce to (v1+v2)/2

[sachin-27]

You're right on the encoding, but reduce doesn't run on this path. The engine aggregates across all shards and returns one single value. Verified this on a 3 node cluster setup on local

[aggarwalmayank]

can we compare this behaviour with vanilla?
When the request has aggregations but the plan produced no rows (empty index, filter matches nothing, etc.), this returns null, so the response omits the aggregations block entirely
whereas i think in vanilla it return the aggregations block with each sub-agg populated with empty/sentinel values (e.g. avg: {"value": null}, sum: {"value": 0.0})

[sachin-27]

So aggResults.isEmpty() tests for empty aggregation plan result. So if request has aggregations, this will never evaluate to true so we will get output in format you are suggesting. Tested this out comparing with Vanilla for a seach matching both docs and both are in sync.

[aggarwalmayank]

For a numeric field like user_id for example

{
"aggs": {
"by_user": {
"terms": { "field": "user_id" }
}
}
}

Where user_id: integer in the mapping.
it will returns StringTerms with string keys.
{ "key": "42", "doc_count": 340 } which will be incorrect right?

shouldn't we detect value source type from mapping and that accordingly generate

[sachin-27]

Pushed commit to fix this

[github-actions]

Persistent review updated to latest commit 2233e09

[github-actions]

Persistent review updated to latest commit b7abe1f

[github-actions]

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