Speed up loading dense singleton keyword fields

[martijnvg]

With this change both sorted set and number doc values use the same bulk loading for values/ordinals.

This PR supersedes (#132715) that also sped up loading dense singleton keyword fields, but duplicated the bulk encoding logic.

[martijnvg]

Any subclasses of BlockLoader.Block aren't accessible from server. So instead of consuming the block that denseOrds would produce, I have a delegate builder that translates the long values to int values (given that singleton ordinals works with int[]).

[elasticsearchmachine]

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

[elasticsearchmachine]

Pinging @elastic/es-storage-engine (Team:StorageEngine)

[martijnvg]

I see the following results when running this query locally: FROM metrics-hostmetricsreceiver.otel-default | STATS count(*) BY host.name | LIMIT 10000. Without this change ~246 ms query time and with this change ~138 ms query time.

When running the following request to get profiling:

POST /_query
{
    "profile": true,
    "query": "FROM metrics-hostmetricsreceiver.otel-default | STATS count(*) BY host.name | LIMIT 10000",
    "pragma": {
        "data_partitioning": "shard"
    }
}

I see the following results:

{
    "operator": "ValuesSourceReaderOperator[fields = [host.name]]",
    "status": {
        "readers_built": {
            "host.name:column_at_a_time:BlockDocValuesReader.SingletonOrdinals": 58
        },
        "values_loaded": 221184000,
        "process_nanos": 1261544847, <-- ~1261ms
        "pages_received": 45597,
        "pages_emitted": 45597,
        "rows_received": 221184000,
        "rows_emitted": 221184000
    }
}

{
    "operator": "ValuesSourceReaderOperator[fields = [host.name]]",
    "status": {
        "readers_built": {
            "host.name:column_at_a_time:BlockDocValuesReader.SingletonOrdinals": 58
        },
        "values_loaded": 221184000,
        "process_nanos": 517397300, <-- ~517ms
        "pages_received": 45597,
        "pages_emitted": 45597,
        "rows_received": 221184000,
        "rows_emitted": 221184000
    }
}

[martijnvg]

The following query: TS metrics-hostmetricsreceiver.otel-default | WHERE metrics.system.memory.utilization IS NOT NULL AND @timestamp >= \"2025-07-30T10:28:00.000Z\" AND @timestamp <= \"2025-07-31T10:28:00.000Z\" | STATS AVG(AVG_OVER_TIME(metrics.system.memory.utilization)) BY host.name, BUCKET(@timestamp, 1h) | LIMIT 10000 execute with a query time of ~170 ms without this change and ~140ms with this change.

[dnhatn]

LGTM. Thanks Martijn!

[martijnvg]

Note that serverless check failed because of the following OOME:

[2025-08-16T05:07:53,058][ERROR][o.e.b.ElasticsearchUncaughtExceptionHandler] [search-ehNQewy] fatal error in thread [elasticsearch[search-ehNQewy][stateless_prewarm][T#18]], exiting
java.lang.OutOfMemoryError: Cannot reserve 2097152 bytes of direct buffer memory (allocated: 266869562, limit: 268435456)
	at java.nio.Bits.reserveMemory(Bits.java:178) ~[?:?]
	at java.nio.DirectByteBuffer.<init>(DirectByteBuffer.java:108) ~[?:?]
	at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:367) ~[?:?]
	at co.elastic.elasticsearch.stateless.cache.SearchCommitPrefetcher.lambda$static$0(SearchCommitPrefetcher.java:75) ~[?:?]
	at java.lang.ThreadLocal$SuppliedThreadLocal.initialValue(ThreadLocal.java:356) ~[?:?]
	at java.lang.ThreadLocal.setInitialValue(ThreadLocal.java:224) ~[?:?]
	at java.lang.ThreadLocal.get(ThreadLocal.java:193) ~[?:?]
	at java.lang.ThreadLocal.get(ThreadLocal.java:171) ~[?:?]
	at co.elastic.elasticsearch.stateless.cache.SearchCommitPrefetcher.lambda$prefetchLatestCommit$7(SearchCommitPrefetcher.java:338) ~[?:?]
	at co.elastic.elasticsearch.stateless.cache.reader.SequentialRangeMissingHandler.lambda$fillCacheRange$0(SequentialRangeMissingHandler.java:211) ~[?:?]
	at org.elasticsearch.action.ActionListenerImplementations$MappedActionListener.lambda$map$0(ActionListenerImplementations.java:124) ~[elasticsearch-9.2.0-SNAPSHOT.jar:?]
	at org.elasticsearch.action.ActionListenerImplementations$MappedActionListener.onResponse(ActionListenerImplementations.java:105) ~[elasticsearch-9.2.0-SNAPSHOT.jar:?]
	at org.elasticsearch.action.ActionRunnable$2.accept(ActionRunnable.java:58) ~[elasticsearch-9.2.0-SNAPSHOT.jar:?]
	at org.elasticsearch.action.ActionRunnable$2.accept(ActionRunnable.java:55) ~[elasticsearch-9.2.0-SNAPSHOT.jar:?]
	at org.elasticsearch.action.ActionRunnable$4.doRun(ActionRunnable.java:101) ~[elasticsearch-9.2.0-SNAPSHOT.jar:?]
	at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:27) ~[elasticsearch-9.2.0-SNAPSHOT.jar:?]
	at org.elasticsearch.common.util.concurrent.EsExecutors$DirectExecutorService.execute(EsExecutors.java:314) ~[elasticsearch-9.2.0-SNAPSHOT.jar:?]
	at co.elastic.elasticsearch.stateless.cache.reader.ObjectStoreCacheBlobReader.getRangeInputStream(ObjectStoreCacheBlobReader.java:67) ~[?:?]
	at co.elastic.elasticsearch.stateless.cache.reader.MeteringCacheBlobReader.getRangeInputStream(MeteringCacheBlobReader.java:51) ~[?:?]
	at co.elastic.elasticsearch.stateless.cache.reader.SequentialRangeMissingHandler.inputStreamFromCacheBlobReader(SequentialRangeMissingHandler.java:242) ~[?:?]
	at co.elastic.elasticsearch.stateless.cache.reader.SequentialRangeMissingHandler.createInputStream(SequentialRangeMissingHandler.java:225) ~[?:?]
	at co.elastic.elasticsearch.stateless.cache.reader.SequentialRangeMissingHandler.fillCacheRange(SequentialRangeMissingHandler.java:208) ~[?:?]
	at org.elasticsearch.blobcache.shared.SharedBlobCacheService$3.fillCacheRange(SharedBlobCacheService.java:698) ~[?:?]
	at org.elasticsearch.blobcache.shared.SharedBlobCacheService$CacheFileRegion.lambda$fillGapRunnable$13(SharedBlobCacheService.java:1165) ~[?:?]
	at org.elasticsearch.action.ActionListener.run(ActionListener.java:465) ~[elasticsearch-9.2.0-SNAPSHOT.jar:?]
	at org.elasticsearch.blobcache.shared.SharedBlobCacheService$CacheFileRegion.lambda$fillGapRunnable$14(SharedBlobCacheService.java:1160) ~[?:?]
	at co.elastic.elasticsearch.stateless.cache.SearchCommitPrefetcher$PrefetchExecutor$1.onResponse(SearchCommitPrefetcher.java:495) ~[?:?]
	at co.elastic.elasticsearch.stateless.cache.SearchCommitPrefetcher$PrefetchExecutor$1.onResponse(SearchCommitPrefetcher.java:491) ~[?:?]
	at org.elasticsearch.common.util.concurrent.AbstractThrottledTaskRunner$1.doRun(AbstractThrottledTaskRunner.java:136) ~[elasticsearch-9.2.0-SNAPSHOT.jar:?]
	at org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingAbstractRunnable.doRun(ThreadContext.java:1067) ~[elasticsearch-9.2.0-SNAPSHOT.jar:?]
	at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:27) ~[elasticsearch-9.2.0-SNAPSHOT.jar:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1095) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:619) ~[?:?]
	at java.lang.Thread.run(Thread.java:1447) ~[?:?]

This is unrelated to this change. But caused all the :qa:stateful:x-pack:plugin:esql:qa:server:single-node:javaRestTest tests to fail.