ES|QL: Make TopN aggregator use heap sort internally.

[przemekwitek]

Until now, the TopN aggregator has been copying the bucket values to the separate array and sorting them using Arrays.sort method call.
This PR makes use of the existing heap structure to sort in-place using standard heap sort algorithm.

[elasticsearchmachine]

Pinging @elastic/es-analytical-engine (Team:Analytics)

[ivancea]

We also have the BytesRefBucketedSort and IpBucketedSort classes that aren't autogenerated. The same comment about the sort is there, if you want to give them a try. I think it should be quite similar, if not identical.

Same with

elasticsearch/server/src/main/java/org/elasticsearch/search/sort/BucketedSort.java

Line 187 in 1ba21c2

// TODO we usually have a heap here so we could use that to build the results sorted.

for _search, which is the base of these classes. I'm not sure if it's worth it there though, and it may be a bit different. @nik9000? (In any case, this one doesn't have to be in this PR, I'm just commenting)

[nik9000]

for _search, which is the base of these classes. I'm not sure if it's worth it there though, and it may be a bit different. @nik9000? (In any case, this one doesn't have to be in this PR, I'm just commenting)

I think the testing is pretty good over there. I'd grab there first and, if you have time, grab the next ones. But these are more imporant I think.

[nik9000]

In BucketedSortTestCase we only use up to three sorted values. It's probably worth adding a case where we use a big list of them here. Just to make sure you didn't off-by-one or round funny or something. Which I'd never catch with code review.

[przemekwitek]

We also have the BytesRefBucketedSort and IpBucketedSort classes that aren't autogenerated. The same comment about the sort is there, if you want to give them a try. I think it should be quite similar, if not identical.

Done.

[przemekwitek]

In BucketedSortTestCase we only use up to three sorted values. It's probably worth adding a case where we use a big list of them here.

IIUC such a test already exists: BucketedSortTestCase.testManyBucketsManyHits.
This test collects 10000 values in random order and then asserts that the values are returned as sorted.

Do you think there is anything more I should do on this PR?
I've noticed some code duplication wrt heap structures between those classes, but that could be better handled in a separate PR if we decide to do so.

[nik9000]

This test collects 10000 values in random order and then asserts that the values are returned as sorted.

Ah. That's good.

I've noticed some code duplication wrt heap structures between those classes, but that could be better handled in a separate PR if we decide to do so.

Yeah. That's a fine for later thing. Some of that duplication often comes from using the templates. Some amount of duplication is to monomorphize the tight loops. We tend to default to this kind of behavior in the hot loop, which this is, and we'll do microbenchmarks if we're trying to be more careful.

[przemekwitek]

Same with

elasticsearch/server/src/main/java/org/elasticsearch/search/sort/BucketedSort.java

Line 187 in 1ba21c2

// TODO we usually have a heap here so we could use that to build the results sorted.

for _search, which is the base of these classes. I'm not sure if it's worth it there though, and it may be a bit different.

Indeed, the code is a bit different, I would say more advanced (it has support for extra values which we may need to port to ES|QL's version at some point).

I did local changes there and run a benchmark.
The version fetched from main has results:

# Warmup Iteration   1: 3.913 ns/op
Iteration   1: 4.008 ns/opING [1m 39s]
Iteration   2: 4.107 ns/opING [1m 49s]
Iteration   3: 4.006 ns/opING [1m 59s]
Iteration   4: 4.163 ns/opING [2m 9s]
Iteration   5: 4.008 ns/opING [2m 19s]
Iteration   6: 4.044 ns/opING [2m 29s]
Iteration   7: 4.101 ns/opING [2m 39s]
Iteration   8: 4.003 ns/opING [2m 49s]
Iteration   9: 4.070 ns/opING [2m 59s]
Iteration  10: 4.147 ns/opING [3m 9s]

Result "org.elasticsearch.benchmark.compute.operator.AggregatorBenchmark.run":
  4.066 ±(99.9%) 0.093 ns/op [Average]
  (min, avg, max) = (4.003, 4.066, 4.163), stdev = 0.061
  CI (99.9%): [3.973, 4.158] (assumes normal distribution)

whereas the local change with heap sort had:

# Warmup Iteration   1: 4.148 ns/op
Iteration   1: 3.989 ns/opING [44s]
Iteration   2: 4.205 ns/opING [54s]
Iteration   3: 4.078 ns/opING [1m 4s]
Iteration   4: 4.300 ns/opING [1m 14s]
Iteration   5: 4.102 ns/opING [1m 24s]
Iteration   6: 3.987 ns/opING [1m 34s]
Iteration   7: 3.933 ns/opING [1m 44s]
Iteration   8: 3.896 ns/opING [1m 54s]
Iteration   9: 3.885 ns/opING [2m 4s]
Iteration  10: 4.241 ns/opING [2m 14s]

Result "org.elasticsearch.benchmark.compute.operator.AggregatorBenchmark.run":
  4.062 ±(99.9%) 0.224 ns/op [Average]
  (min, avg, max) = (3.885, 4.062, 4.300), stdev = 0.148
  CI (99.9%): [3.838, 4.286] (assumes normal distribution)

so it's very similar. The command used was:

./gradlew run -p benchmarks --args "\\.AggregatorBenchmark -pgrouping=none -pop=top -pblockType=vector_longs -pfilter=none -wi 1 -i 10"

[ivancea]

Thanks for the benchies! I suppose it will be difficult to see great improvements in the benchmarks we have, as ingestion will take most of the time and result collection is done only once. Maybe a larger limit and less input data? Anyway, performance didn't suffer apparently, so I think it's nice as it is!

[przemekwitek]

I suppose it will be difficult to see great improvements in the benchmarks we have

FWIW: Apart from the time aspect, we save on some memory as we do not have to allocate this temporary list.