[RFC] Pre Compute Aggregations with Star Tree index

[bharath-techie]

Is your feature request related to a problem? Please describe

Aggregations are the most used query type in observability use cases and the aggregation is typically on metrics, request logs, etc. and spread across different fields. Heavy aggregation queries doesn’t have an upper bound in terms of time taken and resource consumption. In general, the performance of an aggregation query is relative to the number of documents.

Aggregations are faster when they are run on rolled up indices and transforms as they help in reducing granularity and providing materialized views. But, they are generally done once ingestion is complete, similar constructs doesn’t exists for live/active data.

Existing solutions

Aggregation framework

Existing aggregation framework in OpenSearch is quite powerful with bucket, pipeline and metric aggregations. It does support wide range of aggregations since it operates on the non altered live documents of the indices. And as it operates on live original data, deletes and updates are also supported.

While it works great, it has few cons

• Query latency scales linearly with the number of documents.
  • For example, aggregation which measures count of ‘status’ in the HTTP logs workload in OSB, takes a lot longer for '200' status [~200M documents] compared to '400' status [~3000 documents]
• If there are multiple aggregations on different fields, each field / aggregation is processed separately and in general it leads to higher query latencies.
  • The following example captures query time as user adds more aggregations:
    • if user queries on status = 200 + aggregation of sum (size) = ~4s
    • if user queries on status = 200 + aggregations of sum (size) , avg(size) = ~6s
    • if user queries on status = 200 + aggregations of sum (size) , avg(size) , max(size) = ~7.5s

Index rollups / Index transforms

Index rollup jobs lets users reduce data granularity by rolling up old data into condensed indices, transform jobs lets users create a different, summarized view of users' data centered around certain fields. These solutions are generally used to save storage space and also the queries are faster as the number of documents are reduced.

Cons:

• These solutions are generally used to reduce granularity of older data and not used on real time data.
• Configuring the jobs and managing the indices is cumbersome.
  • Cannot search a mix of rollup and non-rollup indexes with the same query
  • Initial rollup job can become expensive depending on the size of the data

Describe the solution you'd like

Star tree index

I’m exploring the use case of pre-computing the aggregations using Star Tree index while indexing the data based on the configured fields (dimensions and metrics) during index creation. This is inspired from http://hanj.cs.illinois.edu/pdf/vldb03_starcube.pdf and Apache Pinot’s Star Tree index. Star Tree helps to enforce upper bound on the aggregation queries ensuring predictable latency and resource usage, it is also storage space efficient and configurable.

Star Tree index is a multi-field index, contrary to existing index structures in Lucene and OpenSearch which are on single field.

Improvements with Star Tree index

• Users can perform faster aggregations with a constant upper bound on query latency.
• Star tree natively supports multi field aggregations.
• Star tree index will be created in real time as part of regular indexing, so the data in star tree will always be up to date with the live data.
• Star tree index consolidates the data and is a storage efficient index, and hence brings following benefits
  • Complex aggregation queries will cause fraction of IO utilization compared to present solutions.
  • Paging also will be more efficient as less data gets paged for every star tree query

While it provides the above improvements, it does have its limitations, they are:

• Star tree index will be created only on append-only indices, updates or deletes will not be supported.
• Star tree index will be used for aggregation queries only if the query input is a subset of the star tree configuration of dimensions and metrics
• Star tree index will support limited set of aggregations such as MIN, MAX, COUNT, SUM, AVG. Others are yet to be explored
• The cardinality of the dimensions should not be very high (like _id fields), otherwise it leads to storage explosion and higher query latencies.
• Changing Star Tree config (dimension or metrics) will require a re-index operation.

Given the above details, it does makes a case on why Star Tree is valuable to OpenSearch, I’m working on prototyping it. I’ll follow up on this issue with prototype details along with the benchmark results.

Related component

Search:Aggregations

Describe alternatives you've considered

Mentioned in the Existing Solutions.

Additional context

No response

[bharath-techie]

Prototype Details

For the prototype, ‘DocValuesFormat’ in Lucene is extended to create Star Tree index with support for ‘Numeric’ fields as the dimensions of the Star Tree. Timestamp field is rounded off to various granularity of epoch such as ‘minute’, ‘hour’, ‘day’ etc. and are also added as dimensions to the Star Tree.

Fig: Star Tree Index Structure

Star Tree indexing

Star tree index structure

• Star Tree index structure as portrayed in the above figure, consists of mainly two parts: the Star Tree and sorted + aggregated documents backed by doc value indices.

• Each node in the Star Tree points to a range of documents.

  • A node is further split into child nodes based on maxLeafDocs configuration.
  • The number of documents a leaf node points to, should be less than or equal to maxLeafDocs. This ensures the maximum number of documents that gets traversed to get to the aggregated value is at most maxLeafDocs, thus providing predictable latencies.

• There are special nodes called ‘star nodes’ which helps in skipping non competitive nodes and also in fetching aggregated document wherever applicable during query time.

• The figure contains three examples explaining the star tree traversal during query: i) Average request size for all requests containing hour 11 and status 200, ii) Count of requests for status = 200 (across all hours, hence hour is set to *) and iii) Average request size of all requests for hour 12 (status is set to *). The examples (ii) and (iii) uses star nodes.

Creation Flow

Star Tree index is created during Lucene ‘flush’ operation.

1. The dimension and metric values are fetched from the segment using the corresponding doc values indices. These records are then sorted by dimension split order and the metrics are aggregated.
2. An initial Star Tree is built over this sorted, summarized data. Further hierarchical splitting and aggregation creates specialized star nodes, enabling pruning and aggregation shortcuts for queries.
3. The Star Tree is then expanded further for dimensions based on ‘maxLeafDocs’ configuration set during index creation.

Merge Flow

1. Merge flow reuses the pre-aggregated Star Tree docs that were created during flush operation.
2. Post that, merge flow repeats the star tree construction process [from step 2 of creation flow]

Lucene index files

• Star Tree index is stored under (new) extension ‘.sti’ part of Lucene segment.
• Individual doc values indices for all the dimensions and metrics of the Star Tree and files are saved under (new) extensions ‘.std’ and ‘.stm

TODO: Create an issue with Lucene on the new formats.

Star Tree query

• A new query and aggregation spec is introduced for Star Tree queries. With this new query, OpenSearch will traverse through the Star Tree index and corresponding (Star Tree) doc values to get the results.
• In Lucene, ‘DocValuesReader’ is extended to read the Star Tree indices and a new method ‘getAggregatedValues’ is exposed in LeafReader to read the Star Tree indices in query time. (TODO: Open issue with Lucene).
• For prototype, since only ‘Numeric’ fields are supported, actual term queries of ‘text’ which involves postings format will be compared against later when text fields support is introduced in Star Tree.
• Star Tree query support can be extended for:
  • Term aggregations and multi term aggregations
  • Date histogram aggregations
  • Range query aggregations
  • Filter aggregations - Filters can be on any number of dimensions in a single query. But across dimensions, only ‘MUST’ will be supported, as filters are applied on one dimension after the next as per the dimensions split order.

POC code

bharath-techie@1d44e1a - OpenSearch code changes

bharath-techie/lucene@c993b9a - Lucene code changes

[bharath-techie]

Benchmark Results

OpenSearch Setup

• Instance type: Amazon EC2 r5.4x.large [16 VCPU and 128 GB of RAM]
• Disk: Amazon EBS volume with 300 GB storage, 125 MiBS throughput and 3000 IOPS
• Number of nodes: 1
• JVM heap: 32 GB

Benchmark Setup

• Instance type: Amazon EC2 r5.4xlarge
• Number of nodes generating load: 1

Workload

• 180 GB of HTTP logs indexed data [around 4 months of data]
• Document distribution based on status: 140k documents of 400s compared to 1 billion documents of 200s.

Star Tree config

Dimensions

1. Status
2. Minute
3. Hour
4. Day
5. Year

Metric

• Count of requests

Search (primarily Aggregations)

Here is a summary of how Star Tree queries fared against the aggregation queries on present day indices. [against Points index mainly ]

Query (aggregation)	Median throughput	50 percentile Service time	Comments
200s in range count	8x better performance	5x faster	3M matching documents
400s in range count	40% improvement	40%	100 matching documents
Total 200s count	1000x better performance (best case scenario)	1000x	Best case scenario [1 billion docs]
Total 400s count	1.8x to 2x improvement	2x improvement	100k matching docs
Hourly histogram	15x improvement	15x improvement

Observations

• Queries in Star Tree have bigger performance gains compared to existing solutions when the number of docs matching the query is large.
• In addition to query performance improvement, there was also significant reduction in page cache utilization for Star Tree queries (30x for 400 count query for example).

Indexing

Given that the aggregation now happens during the indexing, there are tradeoffs w.r.t. storage, indexing throughput, cumulative times during indexing etc. which are captured below.

Metric	Plain indexing (Baseline)	Star Tree indexing (Minutely)	Percentage
Store size	18.47 GB	18.55 GB	0.45%
Cumulative indexing time of primary shards	129.13 min	131.58 min	-1.82%
Median throughput [ Index append ]	348765.32 docs/s	342881.75 docs/s	-1.71%
50% latency [ Index append ]	98.18 ms	98.24 ms	+0.06%
90% latency [ Index append ]	131.40 ms	131.17 ms	-0.17%
99% latency [ Index append ]	513.44 ms	569.45 ms	-9%
Cumulative merge time of primary shards	34.18 min	35.59 min	-3.96%
Cumulative refresh time of primary shards	4.00 min	4.61 min	-13%
Cumulative flush time of primary shards	6.71 min	7.31 min	-8.25%

Observations

• The write threads were occupied on flush flows more during indexing with Star Tree enabled. So there is some impact on the indexing throughput and latency (mainly tail latency) because of that.
• The sort and aggregation part of segment records is quite heavy, hence its done off heap and it scales linearly to number of segment records during flush.
• During merge, the pre-aggregated docs are reused to reconstruct the star tree instead of original docs, hence it is less expensive compared to flush.

Store size with respect to cardinality

Following captures store size with respect to cardinality of timestamp (as timestamp has one of the highest cardinalities and is present in all scenarios)

• Store size without Star Tree - 16.6 GB
• Star Tree index size (sti, std and stm combined)
  • +2 MB - Hourly rounded timestamp [0.01%]
  • +60 MB - Minutely rounded timestamp [0.4 % ]
  • +800 MB - Timestamp ( second ) [ 5% ]

Hence super high cardinality fields must be avoided as part of Star Tree index.

[andrross]

Only a 1000x performance improvement? :) These are awesome results @bharath-techie! One minor note, I'd encourage you to link any prototype code you have on your user fork in case folks want to dive into the implementation.

Star tree index will be created only on append-only indices, updates or deletes will not be supported.

I'm sure there are lots of deep technical issues to dive into here, but I just want to highlight that to my knowledge we don't actually have a true "append-only index" abstraction in OpenSearch today. Data streams are partially that: only "create" operations are allowed when referencing the data stream entity. However, the underlying indexes that make up a data stream are still exposed to users and allow updates and deletes. I'm not sure if that is more of a side effect of the implementation, or if users truly need the option to update/delete documents on a limited basis even for a log analytics-style workload.

[msfroh]

I'm sure there are lots of deep technical issues to dive into here, but I just want to highlight that to my knowledge we don't actually have a true "append-only index" abstraction in OpenSearch today. Data streams are partially that: only "create" operations are allowed when referencing the data stream entity. However, the underlying indexes that make up a data stream are still exposed to users and allow updates and deletes. I'm not sure if that is more of a side effect of the implementation, or if users truly need the option to update/delete documents on a limited basis even for a log analytics-style workload.

We should be able to apply the optimization on a per-segment basis, and fall back to the classic behavior on segments with deletes, right @bharath-techie? Hopefully not all segments would have deletes. Hopefully unmerged deletes would be a transient issue (especially now that it's safe to expunge deletes).

Failing that, I kicked off a thread on lucene-dev to discuss the idea of making deletes cheaply iterable. In that case, we could use the star tree to get the aggregations ignoring deletions, then step through the deleted docs to decrement them from the relevant buckets.

[bharath-techie]

Hi @andrross @msfroh ,
Thanks for the comments.
Once star tree has already pre-computed the aggregations , we go through the star tree index to fetch results for the relevant queries. So document deletions post that are not accounted. Hence results will be inaccurate and that's the current limitation.

We can consider both approaches mentioned by @msfroh to address this gap

• Use traditional aggregations if segment has deletes and use star tree aggregations if otherwise.
  • This will work but it might not be a trivial implementation , so we can probably take this up as a follow up after initial implementation.
• Iterating through deleted docs and negating in the relevant buckets
  • This too works well and only issue I see is cost of recomputing is directly relative to the number of deleted docs. If the number is quite high, then it'll cause slowdowns.
  • We also need to check if certain type of aggregations can be derived with the negated data. Initial supported set ( sum, avg, count, max, min ) can be derived. Other to-be explored aggregations need to be checked on case by case basis.

Updated the POC code links in prototype section.