[PERF][Regression: 3.4 -> 3.6] Composite bucket aggregation is expensive

What is the bug?
There's a significant performance regression in q15, q16, q17, and q19 from OpenSearch-Benchmark's Clickbench queries, originally spotted by @rishabh6788. On my local machine, with 10 million Clickbench records (from OSB's dataset), the following query ran in about 700 ms locally in 3.4, but on mainline is running for ~38 seconds. In OSB itself (with ~100 million records) the queries are simply hitting timeouts and aborting.

How can one reproduce the bug?
Steps to reproduce the behavior:

1. Ingest a few million records from the OSB Clickbench workload (numbers above with 10mil, measurable with less) with the index configuration, I chose these settings for the index:

  "settings": {
    "index.number_of_shards": 1,
    "index.number_of_replicas": 0,
    "index.queries.cache.enabled": false,
    "index.requests.cache.enable": false,
    "index.codec": "best_compression",
    "index.translog.sync_interval": "30s",
    "index.sort.field": [ "CounterID", "EventDate", "UserID", "EventTime", "WatchID" ],
    "index.sort.order": [ "desc", "desc", "desc", "desc", "desc" ]
  },

2. Run this query (q15):

source = clickbench
| where SearchPhrase != ''
| stats count() as c by SearchEngineID, SearchPhrase
| sort - c
| head 10;

3. It's significantly slower than if you do the same steps on 3.4.

What is the expected behavior?

What is your host/environment?

• OS: Linux 5.10, x86_64
• Version: mainline, JDK 25

Do you have any screenshots?
N/A

Do you have any additional context?

Investigation, root cause, and recommendations below

---

Investigation

tl;dr: This is caused by #4884, we fixed the correctness bug and introduced the associated performance problem. Solutions involve trying to find clever ways to avoid needing to do bucket pagination in the first place, but the original benchmark baseline is likely unrecoverable.

For preliminary investigation, I did some measurement. First, checking the plans between the two versions, we see a new TopK physical operator is appearing in mainline queries, as well as several differences in the DSL. (There's leftover quotes due to a copy error, core plan shape is unchanged.) When diffed, the most relevant-looking change is that the composite ordering changed (v3.4 is searchengineid first, mainline is searchphrase first):

mainline:

opensearchsql> explain source = clickbench | where SearchPhrase != '''''' | stats count() as c by SearchEngineID, SearchPhrase | sort - c | head 10;
= Calcite Plan =
== Logical ==
LogicalSystemLimit(sort0=[$0], dir0=[DESC-nulls-last], fetch=[10000], type=[QUERY_SIZE_LIMIT])
  LogicalSort(sort0=[$0], dir0=[DESC-nulls-last], fetch=[10])
    LogicalProject(c=[$2], SearchEngineID=[$0], SearchPhrase=[$1])
      LogicalAggregate(group=[{0, 1}], c=[COUNT()])
        LogicalProject(SearchEngineID=[$64], SearchPhrase=[$62])
          LogicalFilter(condition=[<>($62, '''''')])
            CalciteLogicalIndexScan(table=[[OpenSearch, clickbench]])

== Physical ==
EnumerableLimit(fetch=[10000])
  CalciteEnumerableTopK(sort0=[$0], dir0=[DESC-nulls-last], fetch=[10])
    CalciteEnumerableIndexScan(table=[[OpenSearch, clickbench]], PushDownContext=[[FILTER-><>($0, ''''''), AGGREGATION->rel#957:LogicalAggregate.NONE.[](input=RelSubset#928,group={0, 1},c=COUNT()), PROJECT->[c, SearchEngineID, SearchPhrase]], OpenSearchRequestBuilder(sourceBuilder={
      "from": 0,
      "size": 0,
      "timeout": "1m",
      "query": {
        "bool": {
          "must": [
            {
              "exists": {
                "field": "SearchPhrase",
                "boost": 1.0
              }
            }
          ],
          "must_not": [
            {
              "term": {
                "SearchPhrase": {
                  "value": "''",
                  "boost": 1.0
                }
              }
            }
          ],
          "adjust_pure_negative": true,
          "boost": 1.0
        }
      },
      "aggregations": {
        "composite_buckets": {
          "composite": {
            "size": 10000,
            "sources": [
              {
                "SearchPhrase": {
                  "terms": {
                    "field": "SearchPhrase",
                    "missing_bucket": true,
                    "missing_order": "first",
                    "order": "asc"
                  }
                }
              },
              {
                "SearchEngineID": {
                  "terms": {
                    "field": "SearchEngineID",
                    "missing_bucket": true,
                    "missing_order": "first",
                    "order": "asc"
                  }
                }
              }
            ]
          }
        }
      }
    }, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])

3.4:

= Calcite Plan =
== Logical ==
LogicalSystemLimit(sort0=[$0], dir0=[DESC-nulls-last], fetch=[10000], type=[QUERY_SIZE_LIMIT])
  LogicalSort(sort0=[$0], dir0=[DESC-nulls-last], fetch=[10])
    LogicalProject(c=[$2], SearchEngineID=[$0], SearchPhrase=[$1])
      LogicalAggregate(group=[{0, 1}], c=[COUNT()])
        LogicalProject(SearchEngineID=[$64], SearchPhrase=[$62])
          LogicalFilter(condition=[<>($62, '''''')])
            CalciteLogicalIndexScan(table=[[OpenSearch, clickbench]])

== Physical ==
EnumerableLimit(fetch=[10000])
  EnumerableLimit(fetch=[10])
    EnumerableSort(sort0=[$0], dir0=[DESC-nulls-last])
      CalciteEnumerableIndexScan(table=[[OpenSearch, clickbench]], PushDownContext=[[PROJECT->[SearchPhrase, SearchEngineID], FILTER-><>($0, ''''''), AGGREGATION->rel#353:LogicalAggregate.NONE.[](input=RelSubset#352,group={0, 1},c=COUNT()), PROJECT->[c, SearchEngineID, SearchPhrase]], OpenSearchRequestBuilder(sourceBuilder={
        "from": 0,
        "size": 0,
        "timeout": "1m",
        "query": {
          "bool": {
            "must": [
              {
                "exists": {
                  "field": "SearchPhrase",
                  "boost": 1.0
                }
              }
            ],
            "must_not": [
              {
                "term": {
                  "SearchPhrase": {
                    "value": "''",
                    "boost": 1.0
                  }
                }
              }
            ],
            "adjust_pure_negative": true,
            "boost": 1.0
          }
        },
        "_source": {
          "includes": [
            "SearchPhrase",
            "SearchEngineID"
          ],
          "excludes": []
        },
        "aggregations": {
          "composite_buckets": {
            "composite": {
              "size": 10000,
              "sources": [
                {
                  "SearchEngineID": {
                    "terms": {
                      "field": "SearchEngineID",
                      "missing_bucket": true,
                      "missing_order": "first",
                      "order": "asc"
                    }
                  }
                },
                {
                  "SearchPhrase": {
                    "terms": {
                      "field": "SearchPhrase",
                      "missing_bucket": true,
                      "missing_order": "first",
                      "order": "asc"
                    }
                  }
                }
              ]
            }
          }
        }
      }, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])

In flamegraph measurement, the primary additions are new ActionGet calls which are waiting for a significant amount of time on the underlying cluster (wall clock diff graph, red is where time was added):

This then points at overall degradation for the rest of the cluster, spending more time in CompositeAggregators and DenseConjunctionBulkScorers (CPU graph, red means more CPU here):

I did some measurements of the underlying DSL as well, I measured that the aggregations both produce exactly 881762 buckets. Taking the perf numbers below for the DSL, we get that fetching one page takes ~700ms (consistent with the SQL time), and fetching all pages will take ~60s on the old system and ~40s now (consistent with the SQL time if we're scanning everything).

% hyperfine 'cat /tmp/olddsl.json | xh localhost:9200/clickbench/_search' 'cat /tmp/newdsl.json | xh localhost:9200/clickbench/_search'
Benchmark 1: cat /tmp/olddsl.json | xh localhost:9200/clickbench/_search
  Time (mean ± σ):     749.1 ms ±  19.1 ms    [User: 11.7 ms, System: 3.6 ms]
  Range (min … max):   735.7 ms … 801.9 ms    10 runs

  Warning: The first benchmarking run for this command was significantly slower than the rest (801.9 ms). This could be caused by (filesystem) caches that were not filled until after the first run. You should consider using the '--warmup' option to fill those caches before the actual benchmark. Alternatively, use the '--prepare' option to clear the caches before each timing run.

Benchmark 2: cat /tmp/newdsl.json | xh localhost:9200/clickbench/_search
  Time (mean ± σ):     592.5 ms ±  13.6 ms    [User: 11.1 ms, System: 3.8 ms]
  Range (min … max):   581.4 ms … 616.9 ms    10 runs

Summary
  cat /tmp/newdsl.json | xh localhost:9200/clickbench/_search ran
    1.26 ± 0.04 times faster than cat /tmp/olddsl.json | xh localhost:9200/clickbench/_search

Based on that, the graphs, and some scanning of code diffs between the two, the cause is #4884: we just started paginating composite buckets. So the previous query was fast but incorrect as it was limited to the first page (top 10k) values. This is exactly the expected result, we now return correct results but the associated full index scan takes significantly longer.

A fix would likely look like intelligently leveraging term count sorting (and associated rare sorting where relevant) to avoid needing pagination while maintaining correctness. This is only possible for "simple" aggregation sorts, such as the original query doing an ordinary "sort by count descending & head." If we leverage this, we should gain a good 70-80% improvement, though it's not a silver bullet as OpenSearch's native TopK is still quite intensive:

$ cat /tmp/optimized_dsl.json
{
  "size": 0,
  "timeout": "1m",
  "query": {
    "bool": {
      "must": [
        {
          "exists": {
            "field": "SearchPhrase"
          }
        }
      ],
      "must_not": [
        {
          "term": {
            "SearchPhrase": ""
          }
        }
      ]
    }
  },
  "aggregations": {
    "top_combinations": {
      "multi_terms": {
        "terms": [
          {
            "field": "SearchEngineID"
          },
          {
            "field": "SearchPhrase"
          }
        ],
        "size": 10,
        "order": {
          "_count": "desc"
        }
      }
    }
  }
}

$ hyperfine 'cat /tmp/optimized_dsl.json | xh post localhost:9200/clickbench/_search'
Benchmark 1: cat /tmp/optimized_dsl.json | xh post localhost:9200/clickbench/_search
  Time (mean ± σ):     10.436 s ±  0.035 s    [User: 0.008 s, System: 0.003 s]
  Range (min … max):   10.372 s … 10.482 s    10 runs

---

Additional note: using bucket_nullable=false reduces the current query's time from 38s to ~14.6s (credit: @penghuo)

source = clickbench
| where SearchPhrase != ''
| stats bucket_nullable=false count() as c by SearchEngineID, SearchPhrase
| sort - c
| head 10;

% hyperfine 'cat /tmp/query.ppl | ppl' 'cat /tmp/query_nullable_buckets.ppl | ppl'
Benchmark 1: cat /tmp/query.ppl | ppl
  Time (mean ± σ):     38.344 s ±  0.086 s    [User: 0.101 s, System: 0.031 s]
  Range (min … max):   38.210 s … 38.480 s    10 runs

Benchmark 2: cat /tmp/query_nullable_buckets.ppl | ppl
  Time (mean ± σ):     14.688 s ±  0.042 s    [User: 0.104 s, System: 0.027 s]
  Range (min … max):   14.643 s … 14.768 s    10 runs

Summary
  cat /tmp/query_nullable_buckets.ppl | ppl ran
    2.61 ± 0.01 times faster than cat /tmp/query.ppl | ppl

---

Disabling legacy syntax also makes the queries faster (credit: @rishabh6788)

$ curl -sS -H 'Content-Type: application/json' \
-X PUT localhost:9200/_plugins/_query/settings \
-d '{"transient" : {"plugins.ppl.syntax.legacy.preferred" : "false"}}'

$ hyperfine "$(cat /tmp/query.ppl | ppl --export)" "$(cat /tmp/query2.ppl | ppl --export)"
Benchmark 1: curl -X POST http://localhost:9200/_plugins/_ppl -H 'Content-Type: application/json' -d '{"query": "source = clickbench | where SearchPhrase != '"'"''"'"' | stats count() as c by SearchEngineID, SearchPhrase | sort - c | head 10;"}' --max-time 120
  Time (mean ± σ):     10.418 s ±  0.019 s    [User: 0.002 s, System: 0.008 s]
  Range (min … max):   10.395 s … 10.454 s    10 runs

Benchmark 2: curl -X POST http://localhost:9200/_plugins/_ppl -H 'Content-Type: application/json' -d '{"query": "source = clickbench | where SearchPhrase != '"'"''"'"' | stats bucket_nullable=false count() as c by SearchEngineID, SearchPhrase | sort - c | head 10;"}' --max-time 120
  Time (mean ± σ):     10.404 s ±  0.014 s    [User: 0.007 s, System: 0.004 s]
  Range (min … max):   10.385 s … 10.428 s    10 runs

Summary
  curl -X POST http://localhost:9200/_plugins/_ppl -H 'Content-Type: application/json' -d '{"query": "source = clickbench | where SearchPhrase != '"'"''"'"' | stats bucket_nullable=false count() as c by SearchEngineID, SearchPhrase | sort - c | head 10;"}' --max-time 120 ran
    1.00 ± 0.00 times faster than curl -X POST http://localhost:9200/_plugins/_ppl -H 'Content-Type: application/json' -d '{"query": "source = clickbench | where SearchPhrase != '"'"''"'"' | stats count() as c by SearchEngineID, SearchPhrase | sort - c | head 10;"}' --max-time 120

Reproduce q15 with the OpenSearch-Benchmark Clickbench workload and compare the 3.4 and mainline plans and DSL, focusing on the composite aggregation pagination introduced by #4884. Use the reported composite bucket and term-count sorting measurements as benchmarks; done means preserving correct results while avoiding the regression and timeout behavior.