Support pushdown of ARRAY_AGG (mvcombine) aggregation

[LantaoJin]

The replacement of closed #5072

Description

This PR pushes down the Calcite ARRAY_AGG aggregation to OpenSearch as a top_hits sub-aggregation of the composite aggregation. The PPL mvcombine command produces this aggregation as ARRAY_AGG(x) FILTER (x IS NOT NULL), grouped by every other field. It redoes the closed #5072 on the latest main. Before this change mvcombine always ran in memory: the scan read every matching document through a PIT, even when the output was limited.

#5072 always pushed down, with a top_hits size of plugins.query.buckets (10000 by default). OpenSearch rejects a top_hits larger than the index setting index.max_inner_result_window (default 100). So mvcombine failed on every index that kept the default, and every IT that ran mvcombine needed that setting raised. That is why #5072's ITs never all passed. The setting also guards heap: each shard allocates a hit queue of size slots for every bucket of a composite page. This PR changes the approach:

• Push down only when it is safe. ARRAY_AGG is pushed down only when index.max_inner_result_window is at least plugins.query.buckets for every queried index. Otherwise it stays in memory, so the default behaviour is unchanged. The setting is read lazily, and only for aggregates that contain ARRAY_AGG, by the node client (local and cross-cluster indices) and the REST client. It is cached per OpenSearchIndex.
• Same values as in memory.
  • The target is read from _source, filtered to that one field, and parsed with the field's own type, as the scan does. The fields API would format values differently, for example dates as 2018-06-23T00:00:00.000Z instead of 2018-06-23 00:00:00. Following the review on Support pushdown array/collect aggregation #5072, the full _source is not fetched.
  • One element per hit, so duplicates are kept.
  • A group whose values are all filtered out gives null, as in memory.
  • Alias and expression targets use the fields API and script fields.
• Same order as in memory. mvcombine keeps the order of its input. A sort that was pushed down before the aggregate is now carried into the top_hits sort; the sort digest keeps its sort builders. A sort on an expression prevents the pushdown.
• No silent truncation. If a group has more values than the top_hits size, the query fails with RESOURCE_LIMIT_EXCEEDED and names the settings to raise. OpenSearchExecutionEngine now surfaces an ErrorReport raised while reading results, instead of wrapping it into an opaque error.
• Not pushed down: ARRAY_AGG ... IGNORE NULLS, and object or geo_point targets.

Known limitation: if the target holds an array in a document, the pushed-down mvcombine keeps the array's first value. In memory, the array is stringified, for example "[x, y]". mvcombine requires a single-valued target. The docs state both rules.

Testing:

• Unit tests: AggregateAnalyzerTest, OpenSearchAggregationResponseParserTest, both client tests, OpenSearchIndexTest and OpenSearchExecutionEngineTest.
• CalciteMvCombineCommandIT: compares every result with pushdown on and off, row for row. It covers duplicates, empty strings, a group without values, a preceding sort, and ip and date targets. It also covers the overflow error, and the fallback when the window is too small.
• CalciteExplainIT: the pushed plan, the carried sort, and the fallback for an expression sort. The existing mvcombine explain on a default index is unchanged.
• Full local runs, all passing: ./gradlew build (unit tests and checks); integTest (7483 tests, 269 skipped); integTestWithSecurity (138); yamlRestTest (256); tracingIntegTest (6); doctest (86).

Related Issues

Resolves #5070

Check List

• New functionality includes testing.
• New functionality has been documented.
• New functionality has javadoc added.
• New functionality has a user manual doc added.
• New PPL command checklist all confirmed.
• API changes companion pull request created.
• Commits are signed per the DCO using --signoff or -s.
• Public documentation issue/PR created.

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 🔍

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ Recommended focus areas for review Possible Issue When all remote clusters are skipped but local indices remain, the method returns a RemoteMetadata with empty innerWindows map. This differs from the windows map behavior and may cause issues if code expects innerWindows to be populated for local indices. The caller in getIndexMaxInnerResultWindows does not merge local results when groups.remote() is empty, unlike getIndexMaxResultWindows which does merge them. return new RemoteMappingsReader.RemoteMetadata(Map.of(), Map.of(), Map.of()); Possible Issue When extracting a value from _source using XContentMapValues.extractValue, if the field is missing the method returns null. However, if the field exists but is an empty list, the code returns null (line 80). This conflates "field missing" with "field is empty array", which may not match the in-memory scan behavior for empty arrays versus missing fields. private Object sourceValue(SearchHit hit) { Map<String, Object> source = hit.getSourceAsMap(); Object value = source == null ? null : XContentMapValues.extractValue(sourcePath, source); if (value instanceof List<?> list) { return list.isEmpty() ? null : list.getFirst(); } return value;

[github-actions]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
General	Log index name blocking pushdown When multiple indices are queried with different max_inner_result_window settings, the minimum value is used. If one index has a low setting, pushdown fails for all indices. Consider logging which specific index is blocking the pushdown to aid troubleshooting. opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java [413-425] int maxInnerResultWindow = osIndex.getMaxInnerResultWindow(); if (maxInnerResultWindow < queryBucketSize) { if (LOG.isDebugEnabled()) { LOG.debug( - "Cannot pushdown ARRAY_AGG since index.max_inner_result_window {} is less than the" + "Cannot pushdown ARRAY_AGG since index.max_inner_result_window {} (from index {}) is less than the" + " query bucket size {}", maxInnerResultWindow, + osIndex.getIndexName(), queryBucketSize); } return null; } Suggestion importance[1-10]: 6 __ Why: Adding the index name to the debug log would improve troubleshooting when multiple indices have different max_inner_result_window settings. However, osIndex.getIndexName() may not exist or may return multiple index names, so the implementation needs verification. The suggestion is helpful but requires validation.	Low
	Add buffer to recommended settings The error message suggests increasing settings to totalHits.value(), but this may not be sufficient if the actual group size exceeds this value. Consider recommending a buffer or using a more conservative estimate to prevent repeated failures. opensearch/src/main/java/org/opensearch/sql/opensearch/response/agg/ArrayAggParser.java [44-61] if (totalHits != null && totalHits.value() > hits.length) { + long recommendedSize = totalHits.value() + (totalHits.value() / 10); // Add 10% buffer throw ErrorReport.wrap( new ResourceLimitExceededException( String.format( "Cannot collect all %d values of a group into [%s], only %d were returned.", totalHits.value(), name, hits.length))) .code(ErrorCode.RESOURCE_LIMIT_EXCEEDED) .details( String.format( "The aggregation was pushed down as a top_hits of %d hits per group. Increase" + " both [plugins.query.buckets] and the index setting" - + " [index.max_inner_result_window] to at least %d, or disable the" + + " [index.max_inner_result_window] to at least %d (recommended: %d with buffer), or disable the" + " aggregation pushdown.", - hits.length, totalHits.value())) + hits.length, totalHits.value(), recommendedSize)) .build(); } Suggestion importance[1-10]: 5 __ Why: The suggestion to add a buffer to the recommended settings is reasonable for user experience, but it's not critical. The current implementation correctly identifies the minimum required value (totalHits.value()), and adding a buffer is an enhancement rather than a fix. The score reflects its moderate value as a usability improvement.	Low
	Clarify null handling for empty lists When a multi-valued field contains an empty list, null is returned. However, an empty list and a missing field both result in null, making them indistinguishable. Verify this behavior matches the in-memory implementation to ensure consistency. opensearch/src/main/java/org/opensearch/sql/opensearch/response/agg/ArrayAggParser.java [76-83] private Object sourceValue(SearchHit hit) { Map<String, Object> source = hit.getSourceAsMap(); - Object value = source == null ? null : XContentMapValues.extractValue(sourcePath, source); + if (source == null) { + return null; + } + Object value = XContentMapValues.extractValue(sourcePath, source); if (value instanceof List<?> list) { return list.isEmpty() ? null : list.getFirst(); } return value; } Suggestion importance[1-10]: 3 __ Why: The suggestion's improved_code is nearly identical to the existing_code (only reordering the null check), so it doesn't meaningfully change the behavior. The concern about distinguishing empty lists from missing fields is valid but not addressed by the proposed change. The score is low because the improvement is minimal.	Low

[codecov]

Codecov Report

❌ Patch coverage is 66.01307% with 52 lines in your changes missing coverage. Please review.
✅ Project coverage is 63.89%. Comparing base (4d98393) to head (f5a7a97).
⚠️ Report is 3 commits behind head on main.

Files with missing lines	Patch %	Lines
...ensearch/storage/scan/CalciteLogicalIndexScan.java	0.00%	39 Missing ⚠️
...opensearch/executor/OpenSearchExecutionEngine.java	55.55%	3 Missing and 1 partial ⚠️
...arch/sql/opensearch/request/AggregateAnalyzer.java	90.00%	2 Missing and 1 partial ⚠️
...ch/sql/opensearch/response/agg/ArrayAggParser.java	92.85%	0 Missing and 2 partials ⚠️
...ql/opensearch/storage/scan/context/SortDigest.java	0.00%	2 Missing ⚠️
...ch/sql/opensearch/client/OpenSearchNodeClient.java	92.85%	0 Missing and 1 partial ⚠️
...nsearch/storage/scan/AbstractCalciteIndexScan.java	0.00%	1 Missing ⚠️

❌ Your project check has failed because the head coverage (63.89%) is below the target coverage (99.00%). You can increase the head coverage or adjust the target coverage.

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #5892      +/-   ##
============================================
+ Coverage     63.87%   63.89%   +0.01%     
- Complexity     9091     9119      +28     
============================================
  Files           962      964       +2     
  Lines         41033    41173     +140     
  Branches       4668     4696      +28     
============================================
+ Hits          26211    26308      +97     
- Misses        13944    13982      +38     
- Partials        878      883       +5     

Flag	Coverage Δ
sql-engine	63.89% <66.01%> (+0.01%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.