[META] Advanced Query Engine Performance Optimizations in OpenSearch

[jainankitk]

Overview

This meta issue tracks below complementary performance optimization strategies:

• Skip list based optimizations (Lucene 10.0) for efficient aggregation operations - [META] Skip List Based Optimization for Aggregations in OpenSearch #19384
• Lucene bulk collection API integration (Lucene 10.3/10.4) for vectorized execution and specific aggregation optimizations
• Reducing JVM overhead through strategic call site optimizations and inlining
• Using more recent algorithms/libraries, especially for stats aggregation computations

Key Ideas

Some of the key ideas powering these optimizations based on observations from the recent benchmarks on OpenSearch and interactions with Lucene community are listed below:

Skip List Optimizations

While most of the optimizations based on skip list are covered in the specific issue, adding summarized points here for more holistic picture:

• Optimize multi-filter numeric bucket aggregations using skip lists
• Recent collect(DocIdStream,long) API to improve sub-aggregation bucket aggregation by combining filter rewrite and skip list approach above
• Enhance deleted document handling for filter rewrite based sub aggregation (done as part of Handle deleted documents for filter rewrite subaggregation optimization #19643) and top level aggregations (using SparseFixedBitSet based deleted documents in Lucene 10.4)
• Recent collectRange API to implement per aggregator logic and efficiently skip over large range of documents by leveraging preaggregated data in skiplist

While some optimizations are covered in bulk collection issue, expanding more below:

Bulk Collection API Integration

While skiplist can be leveraged for collectRange API at top level, there is opportunity to adapt specific aggregation implementations for bulk APIs which should provide considerable speedup. Some of the key ideas are below:

• Add couple more bulk collection APIs in LeafBucketCollector
  • collect(int[] doc, long bucket) - This API allows collecting and processing multiple documents into same bucket. Should be able to leverage this API for top level collector or sub aggregation scenarios based on filter rewrite
  • collect(int[] doc, long[] bucket) - This API allows collecting and processing multiple documents across different buckets. While we lose some of the benefits for same bucket, processing this way should still be much faster as the docValues can be loaded and processed in batch, before collecting into different buckets. For example - DateHistogram spends significant time rounding the date values, which can have virtual call overhead due to different implementations of DateTimeRounding. Using the bulk API, all the datetime docValues can be loaded into buffer using bulk API in NumericDocValues and processing the rounded values in a batch
• Aggregation processing using above APIs should be significantly faster as primitive arrays can be vectorized and take better advantage of Panama enhancements going forward
• We have custom DocValues implementation in OpenSearch for dealing with double/float which should also supported bulk API for loading value for multiple documents at once to benefit from bulk collection patterns
• Enabling batch processing capabilities should reduce per-document processing overhead which is one of the biggest bottleneck in aggregation processing framework currently
• We should consider moving to model where multi valued fields are treated as separate types, for leveraging bulk collection optimizations by default

JVM-level Optimizations

Making the switch to bulk collection APIs should help:

• Reduce virtual call site overhead in critical paths
• Implement strategic inlining for vectorized execution
• Optimize call site polymorphism for better JIT compilation
  While these are good optimizations, we currently don't have good benchmarking strategies to quantify these improvements clearly. Enhancing benchmarking methodology ([FEATURE] Add support for polluting call sites before running benchmarks opensearch-benchmark-workloads#709) not only helps measure these improvements more effectively, but also better reflects customer production workloads

Supporting References

Sub META issues:

• Bulk collection API optimization: [META] Use Lucene bulk collection API to speed up aggregation #19324
• Skip list optimization: [META] Skip List Based Optimization for Aggregations in OpenSearch #19384

Issues

Skip list optimization code changes

• Multi filter support Adding logic for histogram aggregation using skiplist #19130
• Combine filter rewrite and skip list: Combining filter rewrite and skip list to optimize sub aggregation #19573
• Handling deleted documents: Handle deleted documents for filter rewrite subaggregation optimization #19643

Percentile aggregation improvement

• Improving the efficiency of percentile aggregation #19622

Virtual call site improvement changes

• Adding bulk collect API for filter rewrite sub aggregation collection #19933
• Allow collectors take advantage of preaggregated data using collectRange API #20009

Next Steps

• Enhance OSB benchmark with call site pollution
• Create proof-of-concept implementations for each optimization
• Measure impact on real-world query patterns
• Best practices for maintaining optimizations

Related component

Search:Performance

[jainankitk]

cc: @asimmahmood1 , @bowenlan-amzn , @peteralfonsi - Please let me know in case I missed something that we are currently looking at, but not covered as part of this META

[peteralfonsi]

We should also take a look at the new DocIdStream.intoArray(), which could be relevant for aggs like sum or avg.

Also, I had spent some time poking around at the virtual call overhead for time rounding in date histograms. I tried a minimal change such that the rounding was done in a static method, so that we didn't have to do lookups based on which DateTimeUnit it was. But weirdly, the virtual calls did not disappear (and the latency actually increased, although histogram latency can be tricky to measure without transferring the exact index structure, so it's probably consistent with "no change"). I have no explanation for this, it was pretty strange. So there might be something else going on here beyond virtual calls at Rounding.DateTimeUnit.roundFloor(). IIRC, those virtual calls showed up even if I only ever did histograms with the same bucket size (like all "minute"), which doesn't make much sense.

[jainankitk]

Thanks @peteralfonsi for the review and adding your feedback.

We should also take a look at the new DocIdStream.intoArray(), which could be relevant for aggs like sum or avg.

Yes, we are planning to take advantage of both DocIdStream.intoArray() and NumericDocValues.longValues(int size, int[] docs, long[] values, long defaultValue) for loading docIds and their corresponding docValues into an array

Regarding the time rounding for date histograms, we can take a look at your code change and associated flame graphs as to why it did not help. But, I have a fairly simple approach in mind:

• Add new api roundFloor(long[] utcMillis, long[] roundedValues)
• Leverage this API on DocIdStream by loading docIds and corresponding docValues into buffer
• Virtual call overhead should atleast be reduced by the size of docIds buffer
• The approach can be generalized for any level of aggregation using the collect(int[] doc, long[] bucket) API