[FEATURE] Push down bucket aggregation with sorting by metrics to `multi-terms` aggregation

[LantaoJin]

Is your feature request related to a problem?
The query

| stats count() as cnt by a, b | sort cnt

will push down to a composite aggregation with 1000 buckets. So the sort by cnt only sorts on the 1000 records. The result is incorrect.

What solution would you like?
Push down such case to multi-terms aggregation https://docs.opensearch.org/latest/aggregations/bucket/multi-terms/

What alternatives have you considered?
A clear and concise description of any alternative solutions or features you've considered.

Do you have any additional context?
Add any other context or screenshots about the feature request here.

[LantaoJin]

For single bucket, it can be pushed to

{
  "aggs": {
    "vi": {
      "terms": {
        "field": "vi",
        "size": 1000,
        "min_doc_count": 1,
        "order": {
          "a": "desc"
        }
      },
      "aggregations": {
        "a": {
          "sum": {
            "field": "age"
          }
        }
      }
    }
  }
}

[penghuo]

@LantaoJin I found Composite aggregation perf could be better then Term aggregation in some case. for example, clickbench Q17,

• Q17-Term run 2635ms,
• Q17-Composite run 1325ms.

### Clear cache in between.
POST {{baseUrl}}/hits/_cache/clear

### Q17 - Term
GET {{baseUrl}}/hits/_search
Content-Type: application/x-ndjson

{
  "size": 0,
  "_source" : false,
  "track_total_hits" : -1,
  "aggs": {
    "user_groups": {
      "terms": {
        "field": "UserID",
        "size": 10
      },
      "aggs": {
        "search_phrase_groups": {
          "terms": {
            "field": "SearchPhrase",
            "size": 10
          }
        }
      }
    }
  }
}

### Q17 - Composite
GET {{baseUrl}}/hits/_search
Content-Type: application/x-ndjson

{
  "size" : 0,
  "_source" : false,
  "track_total_hits" : -1,
  "aggregations" : {
    "groupby" : {
      "composite" : {
        "size" : 10,
        "sources" : [
          {
            "UserIDTerm" : {
              "terms" : {
                "field" : "UserID",
                "missing_bucket" : true,
                "order" : "asc"
              }
            }
          },
          {
            "SearchPhraseTerm" : {
              "terms" : {
                "field" : "SearchPhrase",
                "missing_bucket" : true,
                "order" : "asc"
              }
            }
          }
        ]
      }
    }
  }
}

[LantaoJin]

Here is my testing results (with clearing cache before each query):

1 Terms (one group keys, total 1000 buckets): 20251 ms

{
  "size": 0,
  "_source": false,
  "track_total_hits": -1,
  "aggs": {
    "search_phrase_groups": {
      "terms": {
        "field": "SearchPhrase",
        "size": 1000
      }
    }
  }
}

2 Composite (one group keys, total 1000 buckets): 3798 ms

{
  "size": 0,
  "_source": false,
  "track_total_hits": -1,
  "aggregations": {
    "groupby": {
      "composite": {
        "size": 1000,
        "sources": [
          {
            "SearchPhraseTerm": {
              "terms": {
                "field": "SearchPhrase",
                "missing_bucket": true,
                "order": "asc"
              }
            }
          }
        ]
      }
    }
  }
}

3 Terms (two group keys, high cardinality first, total 100 buckets): 7908 ms

{
  "size": 0,
  "_source" : false,
  "track_total_hits" : -1,
  "aggs": {
    "user_groups": {
      "terms": {
        "field": "UserID",
        "size": 10
      },
      "aggs": {
        "search_phrase_groups": {
          "terms": {
            "field": "SearchPhrase",
            "size": 10
          }
        }
      }
    }
  }
}

4 Composite (two group keys, high cardinality first, total 100 buckets): 4456 ms

{
  "size" : 0,
  "_source" : false,
  "track_total_hits" : -1,
  "aggregations" : {
    "groupby" : {
      "composite" : {
        "size" : 100,
        "sources" : [
          {
            "UserIDTerm" : {
              "terms" : {
                "field" : "UserID",
                "missing_bucket" : true,
                "order" : "asc"
              }
            }
          },
          {
            "SearchPhraseTerm" : {
              "terms" : {
                "field" : "SearchPhrase",
                "missing_bucket" : true,
                "order" : "asc"
              }
            }
          }
        ]
      }
    }
  }
}

5 Multi-Terms (two group keys, high cardinality first, total 100 buckets): 66168 ms

{
  "size": 0, 
  "aggs": {
    "hot": {
      "multi_terms": {
        "terms": [{
          "field": "UserID" 
        },{
          "field": "SearchPhrase" 
        }],
        "size" : 100
      }    
    }
  }
}

6 Date Histogram (one group key): 4 ms

{
  "size": 0,
  "_source" : false,
  "track_total_hits" : -1,
  "aggs": {
    "user_groups": {
      "date_histogram": {
        "field": "EventDate",
        "interval": "hour"
      }
    }
  }
}

7 Date Histogram with Composite (one group key, 1000 buckets): 1984 ms

{
  "size" : 0,
  "_source" : false,
  "track_total_hits" : -1,
  "aggregations" : {
    "groupby" : {
      "composite" : {
        "size" : 1000,
        "sources": [
          {
            "span(@timestamp,1h)": {
              "date_histogram": {
                "field": "EventDate",
                "missing_bucket": true,
                "missing_order": "first",
                "order": "asc",
                "fixed_interval": "1h"
              }
            }
          }
        ]
      }
    }
  }
}

8 Date Histogram + Terms (two group keys, unlimited * 10 buckets): 5930 ms

{
  "size": 0,
  "_source" : false,
  "track_total_hits" : -1,
  "aggs": {
    "user_groups": {
      "date_histogram": {
        "field": "EventDate",
        "fixed_interval": "1h"
      },
      "aggs": {
        "search_phrase_groups": {
          "terms": {
            "field": "SearchPhrase",
            "size": 10
          }
        }
      }
    }
  }
}

9 Date Histogram with Composite (two group keys, 1000 buckets): 4581 ms

{
  "size" : 0,
  "_source" : false,
  "track_total_hits" : -1,
  "aggregations" : {
    "groupby" : {
      "composite" : {
        "size" : 1000,
        "sources" : [
          {
            "span(@timestamp,1h)": {
              "date_histogram": {
                "field": "EventDate",
                "missing_bucket": true,
                "missing_order": "first",
                "order": "asc",
                "fixed_interval": "1h"
              }
            }
          },
          {
            "SearchPhraseTerm" : {
              "terms" : {
                "field" : "SearchPhrase",
                "missing_bucket" : true,
                "order" : "asc"
              }
            }
          }
        ]
      }
    }
  }
}

10 Histogram (one group key): 1499 ms

{
  "size": 0,
  "_source" : false,
  "track_total_hits" : -1,
  "aggs": {
    "user_groups": {
      "histogram": {
        "field": "UserID",
        "interval": 1000000000000000
      }
    }
  }
}

11 Histogram with Composite(one group key): 1794 ms

{
  "size" : 0,
  "_source" : false,
  "track_total_hits" : -1,
  "aggregations" : {
    "groupby" : {
      "composite" : {
        "size" : 1000,
        "sources": [
          {
            "span(UserID)": {
              "histogram": {
                "field": "UserID",
                "missing_bucket": true,
                "missing_order": "first",
                "order": "asc",
                "interval": 1000000000000000
              }
            }
          }
        ]
      }
    }
  }
}

Summary:

• Data Histogram (6) is much faster than Data Histogram with Composite (7). (6 >> 7).
• Composite Aggregation has better performance in other cases. (2 >> 1, 4 >> 3, 9 >> 8, 10 ≈ 11).
• Multi-Terms Aggregation (5) is worst.

[LantaoJin]

Based on above conclusion, we should make the follow decision:

• When bucket nullable is false, and group-by one data histogram field, pushdown to data histogram aggregation.
  • Partially revert the merged PR of [BUG]Span query in PPL is slower than date histogram aggregation in query DSL #3528
• For group-by then sort by metrics case, sort metrics in memory (max 65535 * 65535 rows) would be better than pushdown multi-terms?
  • If [RFC] Paginating composite aggregation or pushdown size_limit to bucket size #4304 merged, should we pushdown to multi-terms?
• For the rest -> composite aggregation.
  • Close [FEATURE] PPL should support terms aggregation with multiple buckets #4208 as won't fix

cc @penghuo

[penghuo]

Remove missing_bucket and missing_order, Histogram composite aggregation can benefit from range-query rewrite and deliver performance comparable to terms aggregation. e.g.

{
  "profile": true,
  "size" : 0,
  "_source" : false,
  "track_total_hits" : -1,
  "aggregations" : {
    "groupby" : {
      "composite" : {
        "size" : 1000,
        "sources": [
          {
            "span(@timestamp,1h)": {
              "date_histogram": {
                "field": "EventTime",
                "fixed_interval": "1h"
              }
            }
          }
        ]
      }
    }
  }
}

FYI, https://opensearch.org/blog/transforming-bucket-aggregations-our-journey-to-100x-performance-improvement/