Understand/Improve the performance of date histograms

[jainankitk]

I have been looking at understanding and improving the performance of date histograms in Opensearch. For this purpose, I have setup single node cluster, and ingested nyc_taxis dataset.

While running single histogram aggregation request in loop, I collected the following cpu flamegraph:

Trying to understand how can we reduce the scoreRange cost for this request

[jainankitk]

I did some more experiments running histogram query on one node setup and tracking the took times:

$ curl -s -X GET "localhost:9200/nyc_taxis/_search?pretty" -H 'Content-Type: application/json' -d'{"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-01-01 00:00:00","lt": "2016-01-01 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}' | head
{
  "took" : 8822,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {

Initially, I rewrote this as cardinality filter query and took time was significantly lesser for one month:

$ curl -s -X GET "localhost:9200/nyc_taxis/_search?pretty" -H 'Content-Type: application/json' -d'{"size": 0,"track_total_hits": "true","query": {"range": {"dropoff_datetime": {"gte": "2015-12-01 00:00:00","lt": "2015-12-31 00:00:00"}}}}' | head
{
  "took" : 162,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {

Even though the took for one month is much lesser (and overall query latency can be significantly reduced), I was not too keen on this approach as it had very limited applications (only cardinality aggregations without aggregation composition)

[jainankitk]

While going through the lucene code, I noticed that the critical path is date histogram collector which is very iterative in nature. But, it is very efficient at computing results for many levels of composition with single iteration. Hence, I came up with another way of rewriting the query applicable to more histogram aggregation queries with following properties:

• Leading scorer is date range filter
• Top level aggregation is date range

Key idea is to slice the date range itself into multiple queries using rewrite. The response transformation is quite simple - union over disjoint sets. Sharing some results below (doing the rewrite on client instead of server):

Partitioning into 2 slices:

$ curl -s -X GET "localhost:9200/nyc_taxis/_msearch?pretty" -H 'Content-Type: application/json' -d'
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-01-01 00:00:00","lt": "2015-06-30 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-07-01 00:00:00","lt": "2015-12-31 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}
> ' | head
{
  "took" : 4225,
  "responses" : [
    {
      "took" : 4224,
      "timed_out" : false,
      "_shards" : {
        "total" : 1,
        "successful" : 1,
        "skipped" : 0,

Partitioning into 4 slices:

$ curl -s -X GET "localhost:9200/nyc_taxis/_msearch?pretty" -H 'Content-Type: application/json' -d'
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-01-01 00:00:00","lt": "2015-03-31 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-04-01 00:00:00","lt": "2015-06-30 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-07-01 00:00:00","lt": "2015-09-30 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-10-01 00:00:00","lt": "2015-12-31 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}
> '  | head
{
  "took" : 3588,
  "responses" : [
    {
      "took" : 3583,
      "timed_out" : false,
      "_shards" : {
        "total" : 1,
        "successful" : 1,
        "skipped" : 0,

Partitioning into 12 slices:

$ curl -s -X GET "localhost:9200/nyc_taxis/_msearch?pretty" -H 'Content-Type: application/json' -d'
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-01-01 00:00:00","lt": "2015-01-31 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-02-01 00:00:00","lt": "2015-02-28 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-03-01 00:00:00","lt": "2015-03-31 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-04-01 00:00:00","lt": "2015-04-30 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-05-01 00:00:00","lt": "2015-05-31 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-06-01 00:00:00","lt": "2015-06-30 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-07-01 00:00:00","lt": "2015-07-31 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-08-01 00:00:00","lt": "2015-08-31 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-09-01 00:00:00","lt": "2015-09-30 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-10-01 00:00:00","lt": "2015-10-31 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-11-01 00:00:00","lt": "2015-11-30 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-12-01 00:00:00","lt": "2015-12-31 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"}}}}
> ' | head
{
  "took" : 3291,
  "responses" : [
    {
      "took" : 1813,
      "timed_out" : false,
      "_shards" : {
        "total" : 1,
        "successful" : 1,
        "skipped" : 0,

[jainankitk]

As we can observe from took times above, the most reduction comes from 1 to 2 slices (~50%), increasing the number of slices further does not see as much jump. The limitation from single node compute might also be coming into play

If we combine the slicing approach and cardinality specific rewrite, the reduction is more than 80% in latency. Although, that approach does not give any flexibility on the number of slices and limited to specific cardinality query like above. Sample invocation:

$ curl -s -X GET "localhost:9200/nyc_taxis/_msearch?pretty" -H 'Content-Type: application/json' -d'
> {}
> {"size": 0,"track_total_hits": "true","query": {"range": {"dropoff_datetime": {"gte": "2015-01-01 00:00:00","lt": "2015-01-31 00:00:00"}}}}
> {}
> {"size": 0,"track_total_hits": "true","query": {"range": {"dropoff_datetime": {"gte": "2015-02-01 00:00:00","lt": "2015-02-28 00:00:00"}}}}
> {}
> {"size": 0,"track_total_hits": "true","query": {"range": {"dropoff_datetime": {"gte": "2015-03-01 00:00:00","lt": "2015-03-31 00:00:00"}}}}
> {}
> {"size": 0,"track_total_hits": "true","query": {"range": {"dropoff_datetime": {"gte": "2015-04-01 00:00:00","lt": "2015-04-30 00:00:00"}}}}
> {}
> {"size": 0,"track_total_hits": "true","query": {"range": {"dropoff_datetime": {"gte": "2015-05-01 00:00:00","lt": "2015-05-31 00:00:00"}}}}
> {}
> {"size": 0,"track_total_hits": "true","query": {"range": {"dropoff_datetime": {"gte": "2015-06-01 00:00:00","lt": "2015-06-30 00:00:00"}}}}
> {}
> {"size": 0,"track_total_hits": "true","query": {"range": {"dropoff_datetime": {"gte": "2015-07-01 00:00:00","lt": "2015-07-31 00:00:00"}}}}
> {}
> {"size": 0,"track_total_hits": "true","query": {"range": {"dropoff_datetime": {"gte": "2015-08-01 00:00:00","lt": "2015-08-31 00:00:00"}}}}
> {}
> {"size": 0,"track_total_hits": "true","query": {"range": {"dropoff_datetime": {"gte": "2015-09-01 00:00:00","lt": "2015-09-30 00:00:00"}}}}
> {}
> {"size": 0,"track_total_hits": "true","query": {"range": {"dropoff_datetime": {"gte": "2015-10-01 00:00:00","lt": "2015-10-31 00:00:00"}}}}
> {}
> {"size": 0,"track_total_hits": "true","query": {"range": {"dropoff_datetime": {"gte": "2015-11-01 00:00:00","lt": "2015-11-30 00:00:00"}}}}
> {}
> {"size": 0,"track_total_hits": "true","query": {"range": {"dropoff_datetime": {"gte": "2015-12-01 00:00:00","lt": "2015-12-31 00:00:00"}}}}
> ' | head
{
  "took" : 993,
  "responses" : [
    {
      "took" : 551,
      "timed_out" : false,
      "_shards" : {
        "total" : 1,
        "successful" : 1,
        "skipped" : 0,

[jainankitk]

Applying the same slicing concept on composite date histogram aggregation, we can see similar improvement (26s to 14s):

$ curl -s -X GET "localhost:9200/nyc_taxis/_search?pretty" -H 'Content-Type: application/json' -d'{"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-01-01 00:00:00","lt": "2016-01-01 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"},"aggs":{"total_amount":{"stats":{"field":"total_amount"}},"tip_amount":{"stats":{"field":"tip_amount"}},"trip_distance":{"stats":{"field":"trip_distance"}}}}}}' | head
{
  "took" : 26604,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {

$ curl -s -X GET "localhost:9200/nyc_taxis/_msearch?pretty" -H 'Content-Type: application/json' -d'
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-01-01 00:00:00","lt": "2015-07-01 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"},"aggs":{"total_amount":{"stats":{"field":"total_amount"}},"tip_amount":{"stats":{"field":"tip_amount"}},"trip_distance":{"stats":{"field":"trip_distance"}}}}}}
> {}
> {"size": 0,"query": {"range": {"dropoff_datetime": {"gte": "2015-07-01 00:00:00","lt": "2016-01-01 00:00:00"}}},"aggs": {"dropoffs_over_time": {"date_histogram": {"field": "dropoff_datetime","calendar_interval": "month"},"aggs":{"total_amount":{"stats":{"field":"total_amount"}},"tip_amount":{"stats":{"field":"tip_amount"}},"trip_distance":{"stats":{"field":"trip_distance"}}}}}}
> ' | head
{
  "took" : 14806,
  "responses" : [
    {
      "took" : 14806,
      "timed_out" : false,
      "_shards" : {
        "total" : 1,
        "successful" : 1,
        "skipped" : 0,

[backslasht]

Thanks @jainankitk for the details.

While this approach of split (into multiple smaller range queries) and merge (into one result) looks promising, I think the net CPU utilization will remain the same. Or in other words, in a busy system where are all search threads are (nearly) occupied, the performance benefit may not be seen.

Also, is there any overhead caused by the number of splits, for example, consider a date aggregation query where the results are aggregated at a month level for an year and say the data is split across 20 shards in 20 different nodes. In that case, instead of 20 network call (between the nodes), this would make 20*12 network calls and there may be some overhead with respect of serializing/de-serializing the request/responses.

Have you tried this with a setup containing more number of nodes?

[msfroh]

@jainankitk How do you think this would compare to using concurrent search across two slices? Unless I'm mistaken, the drop in latency comes from running the two slices in parallel in the _msearch versus running them on a single thread per shard.

Using concurrent search, we would mostly end up doing the same thing, especially if segments hold disjoint contiguous time ranges (which isn't guaranteed, but can be made more likely with #9241).

[jainankitk]

@jainankitk How do you think this would compare to using concurrent search across two slices?

The impact will be similar.

Using concurrent search, we would mostly end up doing the same thing, especially if segments hold disjoint contiguous time ranges (which isn't guaranteed, but can be made more likely with #9241).

While the benefits are similar, this approach is more generic (for date histogram aggregations) as it is agnostic of number of segments. Also, the rewrite using track_total_hits (for cardinality date histogram aggregtions) makes the request scorer bound instead of collector, which is significant improvement

[jainankitk]

While this approach of split (into multiple smaller range queries) and merge (into one result) looks promising, I think the net CPU utilization will remain the same. Or in other words, in a busy system where are all search threads are (nearly) occupied, the performance benefit may not be seen.

@backslasht - Completely agree with you on this. Although I don't have any data on this, IMO date histogram aggregations are mostly generated by UI or dashboard components for visualizing the data. They are quite expensive as well (~7s) rendering them too slow or impractical for low latency applications. Hence, the request concurrency should not be high.

Also, is there any overhead caused by the number of splits, for example, consider a date aggregation query where the results are aggregated at a month level for an year and say the data is split across 20 shards in 20 different nodes. In that case, instead of 20 network call (between the nodes), this would make 20*12 network calls and there may be some overhead with respect of serializing/de-serializing the request/responses.

Assuming the throughput of such requests is not high, the overhead caused due to 20*12 network calls should not be huge. We can also evaluate the query splitting on data node (instead of coordinator) to avoid network calls overhead.

[gashutos]

@jainankitk how many number of shards you have in index ? Running your query with 2 slice is equivalent to doubling the shard count right ?
I think this will give even better improvement on top of concurrent segment search. with higher number of replicas since we will able to achieve more parallelization.

[jainankitk]

@jainankitk how many number of shards you have in index ?

I ran this experiment with single shard, single node.

Running your query with 2 slice is equivalent to doubling the shard count right ?

Yes, but doubling the shard count dynamically is not possible. It already is what it is. Given any shard count, we can always rewrite to slice the query like above

I think this will give even better improvement on top of concurrent segment search. with higher number of replicas since we will able to achieve more parallelization.

While I don't understand the correlation with higher number of replicas, the rewriting is additive to concurrent segment search, given enough CPU is available

[gashutos]

While I don't understand the correlation with higher number of replicas, the rewriting is additive to concurrent segment search, given enough CPU is available

Let say if we have 3 replica for a primary shard, we will have 4 shards total after all on 4 nodes.
Search request for that shard will land to any shard selected through ARS ( adaptive replica selection ). So 4 concurrent search request will be served through 4 different nodes and all individual replica will utilize available individual node CPU through concurrent segment search.
So given if we rewrite query to act like 4 different parallel query, it is latency wise benefit on top of concurrent search.
But here results wont be consistent since replicas might have different documents based on sync state with primary