Push compute engine value loading for dense singleton ordinals down to tsdb codec.

This change targets reading field values in bulk mode at codec level when doc values type is sorted doc values or sorted set doc values, there is only one value per document, and the field is dense (all documents have a value).

Relates to #128445

Author: martijnvg

server/src/main/java/org/elasticsearch/index/codec/tsdb/es819/BulkNumericDocValues.java

@@ -24,4 +24,7 @@ public abstract class BulkNumericDocValues extends NumericDocValues {
      */
     public abstract BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset) throws IOException;
 
+    public abstract BlockLoader.Block readOrdinals(BlockLoader.SingletonOrdinalsBuilder builder, BlockLoader.Docs docs, int offset)
+        throws IOException;
+
 }

server/src/main/java/org/elasticsearch/index/codec/tsdb/es819/BulkSortedDocValues.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.index.codec.tsdb.es819;
+
+import org.apache.lucene.index.SortedDocValues;
+import org.elasticsearch.index.mapper.BlockLoader;
+
+import java.io.IOException;
+
+public abstract class BulkSortedDocValues extends SortedDocValues {
+
+    public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
+    public boolean supportsBlockRead() {
+        return false;
+    }
+
+}

server/src/main/java/org/elasticsearch/index/codec/tsdb/es819/ES819TSDBDocValuesProducer.java

@@ -369,10 +369,26 @@ public int advance(int target) throws IOException {
             public long cost() {
                 return ords.cost();
             }
+
+            @Override
+            public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset) throws IOException {
+                if (ords instanceof BulkNumericDocValues b) {
+                    try (var builder = factory.singletonOrdinalsBuilder(this, docs.count() - offset)) {
+                        return b.readOrdinals(builder, docs, offset);
+                    }
+                } else {
+                    throw new UnsupportedOperationException();
+                }
+            }
+
+            @Override
+            public boolean supportsBlockRead() {
+                return ords instanceof BulkNumericDocValues;
+            }
         };
     }
 
-    abstract class BaseSortedDocValues extends SortedDocValues {
+    abstract class BaseSortedDocValues extends BulkSortedDocValues {
 
         final SortedEntry entry;
         final TermsEnum termsEnum;
@@ -1238,6 +1254,59 @@ public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs
                     }
                 }
 
+                @Override
+                public BlockLoader.Block readOrdinals(BlockLoader.SingletonOrdinalsBuilder builder, BlockLoader.Docs docs, int offset)
+                    throws IOException {
+                    assert maxOrd >= 0 : "unexpected maxOrd[" + maxOrd + "]";
+                    final int docsCount = docs.count();
+                    doc = docs.get(docsCount - 1);
+                    for (int i = offset; i < docsCount;) {
+                        int index = docs.get(i);
+                        final int blockIndex = index >>> ES819TSDBDocValuesFormat.NUMERIC_BLOCK_SHIFT;
+                        final int blockInIndex = index & ES819TSDBDocValuesFormat.NUMERIC_BLOCK_MASK;
+                        if (blockIndex != currentBlockIndex) {
+                            assert blockIndex > currentBlockIndex : blockIndex + " < " + currentBlockIndex;
+                            // no need to seek if the loading block is the next block
+                            if (currentBlockIndex + 1 != blockIndex) {
+                                valuesData.seek(indexReader.get(blockIndex));
+                            }
+                            currentBlockIndex = blockIndex;
+                            decoder.decodeOrdinals(valuesData, currentBlock, bitsPerOrd);
+                        }
+
+                        // Try to append more than just one value:
+                        // Instead of iterating over docs and find the max length, take an optimistic approach to avoid as
+                        // many comparisons as there are remaining docs and instead do at most 7 comparisons:
+                        int length = 1;
+                        int remainingBlockLength = Math.min(ES819TSDBDocValuesFormat.NUMERIC_BLOCK_SIZE - blockInIndex, docsCount - i);
+                        for (int newLength = remainingBlockLength; newLength > 1; newLength = newLength >> 1) {
+                            int lastIndex = i + newLength - 1;
+                            if (isDense(index, docs.get(lastIndex), newLength)) {
+                                length = newLength;
+                                break;
+                            }
+                        }
+
+                        // Unfortunately, no array copy here...
+                        // Since we need to loop here, let's also keep track of min/max.
+                        int minOrd = Integer.MAX_VALUE;
+                        int maxOrd = Integer.MIN_VALUE;
+                        int counter = 0;
+                        int[] convertedOrds = new int[length];
+                        int endBlockInIndex = blockInIndex + length;
+                        for (int j = blockInIndex; j < endBlockInIndex; j++) {
+                            int ord = Math.toIntExact(currentBlock[j]);
+                            convertedOrds[counter++] = ord;
+                            minOrd = Math.min(minOrd, ord);
+                            maxOrd = Math.max(maxOrd, ord);
+                        }
+                        builder.appendOrds(convertedOrds, 0, length, minOrd, maxOrd);
+                        i += length;
+                    }
+                    return builder.build();
+
+                }
+
                 static boolean isDense(int firstDocId, int lastDocId, int length) {
                     // This does not detect duplicate docids (e.g [1, 1, 2, 4] would be detected as dense),
                     // this can happen with enrich or lookup. However this codec isn't used for enrich / lookup.

server/src/main/java/org/elasticsearch/index/mapper/BlockDocValuesReader.java

@@ -22,6 +22,7 @@
 import org.elasticsearch.common.io.stream.ByteArrayStreamInput;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.codec.tsdb.es819.BulkNumericDocValues;
+import org.elasticsearch.index.codec.tsdb.es819.BulkSortedDocValues;
 import org.elasticsearch.index.mapper.BlockLoader.BlockFactory;
 import org.elasticsearch.index.mapper.BlockLoader.BooleanBuilder;
 import org.elasticsearch.index.mapper.BlockLoader.Builder;
@@ -659,6 +660,9 @@ public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throw
             if (docs.count() - offset == 1) {
                 return readSingleDoc(factory, docs.get(offset));
             }
+            if (ordinals instanceof BulkSortedDocValues bulkDv && bulkDv.supportsBlockRead()) {
+                return bulkDv.read(factory, docs, offset);
+            }
             try (var builder = factory.singletonOrdinalsBuilder(ordinals, docs.count() - offset)) {
                 for (int i = offset; i < docs.count(); i++) {
                     int doc = docs.get(i);

server/src/main/java/org/elasticsearch/index/mapper/BlockLoader.java

@@ -531,6 +531,8 @@ interface SingletonOrdinalsBuilder extends Builder {
          * Appends an ordinal to the builder.
          */
         SingletonOrdinalsBuilder appendOrd(int value);
+
+        SingletonOrdinalsBuilder appendOrds(int[] values, int from, int length, int minOrd, int maxOrd);
     }
 
     interface SortedSetOrdinalsBuilder extends Builder {

test/framework/src/main/java/org/elasticsearch/index/mapper/TestBlock.java

@@ -283,6 +283,14 @@ public SingletonOrdsBuilder appendOrd(int value) {
                             throw new UncheckedIOException(e);
                         }
                     }
+
+                    @Override
+                    public BlockLoader.SingletonOrdinalsBuilder appendOrds(int[] values, int from, int length, int minOrd, int maxOrd) {
+                        for (int i = from; i < from + length; i++) {
+                            appendOrd(values[i]);
+                        }
+                        return this;
+                    }
                 }
                 return new SingletonOrdsBuilder();
             }

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/lucene/read/SingletonOrdinalsBuilder.java

@@ -55,6 +55,15 @@ public SingletonOrdinalsBuilder appendOrd(int ord) {
         return this;
     }
 
+    @Override
+    public BlockLoader.SingletonOrdinalsBuilder appendOrds(int[] values, int from, int length, int minOrd, int maxOrd) {
+        System.arraycopy(values, from, ords, count, length);
+        this.count += length;
+        this.minOrd = Math.min(this.minOrd, minOrd);
+        this.maxOrd = Math.max(this.maxOrd, maxOrd);
+        return this;
+    }
+
     @Override
     public SingletonOrdinalsBuilder beginPositionEntry() {
         throw new UnsupportedOperationException("should only have one value per doc");