Introduce index.mapping.use_binary_doc_values to experiment with binary doc values for keyword field.

benchmarks/src/main/java/org/elasticsearch/benchmark/_nightly/esql/ValuesSourceReaderBenchmark.java

@@ -222,7 +222,8 @@ private static BlockLoader blockLoader(String name) {
                 Lucene.KEYWORD_ANALYZER,
                 Lucene.KEYWORD_ANALYZER,
                 new KeywordFieldMapper.Builder(name, IndexVersion.current()).docValues(ft.docValuesType() != DocValuesType.NONE),
-                syntheticSource
+                syntheticSource,
+                false
             ).blockLoader(new MappedFieldType.BlockLoaderContext() {
                 @Override
                 public String indexName() {

modules/mapper-extras/src/test/java/org/elasticsearch/index/mapper/extras/MatchOnlyTextFieldTypeTests.java

@@ -298,7 +298,8 @@ public void testBlockLoaderDoesNotUseSyntheticSourceDelegateWhenIgnoreAboveIsSet
             mock(NamedAnalyzer.class),
             mock(NamedAnalyzer.class),
             builder,
-            true
+            true,
+            false
         );
 
         MatchOnlyTextFieldMapper.MatchOnlyTextFieldType ft = new MatchOnlyTextFieldMapper.MatchOnlyTextFieldType(
@@ -346,7 +347,8 @@ public void testBlockLoaderDoesNotUseSyntheticSourceDelegateWhenIgnoreAboveIsSet
             mock(NamedAnalyzer.class),
             mock(NamedAnalyzer.class),
             builder,
-            true
+            true,
+            false
         );
 
         MatchOnlyTextFieldMapper.MatchOnlyTextFieldType ft = new MatchOnlyTextFieldMapper.MatchOnlyTextFieldType(

server/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java

@@ -243,6 +243,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
         if (IndexSettings.DOC_VALUES_SKIPPER) {
             settings.add(IndexSettings.USE_DOC_VALUES_SKIPPER);
         }
+        settings.add(IndexSettings.USE_BINARY_DOC_VALUES);
         settings.add(IndexSettings.INDEX_MAPPING_EXCLUDE_SOURCE_VECTORS_SETTING);
         BUILT_IN_INDEX_SETTINGS = Collections.unmodifiableSet(settings);
     };

server/src/main/java/org/elasticsearch/index/IndexSettings.java

@@ -675,6 +675,13 @@ public boolean isES87TSDBCodecEnabled() {
         Property.Final
     );
 
+    public static final Setting<Boolean> USE_BINARY_DOC_VALUES = Setting.boolSetting(
+        "index.mapping.use_binary_doc_values",
+        false,
+        Property.IndexScope,
+        Property.Final
+    );
+
     /**
      * The {@link IndexMode "mode"} of the index.
      */

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

@@ -208,6 +208,26 @@ public BytesRef binaryValue() throws IOException {
                         bytesSlice.readBytes((long) doc * length, bytes.bytes, 0, length);
                         return bytes;
                     }
+
+                    @Override
+                    public BlockLoader.Block tryRead(
+                        BlockLoader.BlockFactory factory,
+                        BlockLoader.Docs docs,
+                        int offset,
+                        boolean nullsFiltered,
+                        BlockDocValuesReader.ToDouble toDouble,
+                        boolean toInt
+                    ) throws IOException {
+                        int count = docs.count() - offset;
+                        try (var builder = factory.bytesRefs(count)) {
+                            for (int i = offset; i < docs.count(); i++) {
+                                doc = docs.get(i);
+                                bytesSlice.readBytes((long) doc * length, bytes.bytes, 0, length);
+                                builder.appendBytesRef(bytes);
+                            }
+                            return builder.build();
+                        }
+                    }
                 };
             } else {
                 // variable length
@@ -223,6 +243,28 @@ public BytesRef binaryValue() throws IOException {
                         bytesSlice.readBytes(startOffset, bytes.bytes, 0, bytes.length);
                         return bytes;
                     }
+
+                    @Override
+                    public BlockLoader.Block tryRead(
+                        BlockLoader.BlockFactory factory,
+                        BlockLoader.Docs docs,
+                        int offset,
+                        boolean nullsFiltered,
+                        BlockDocValuesReader.ToDouble toDouble,
+                        boolean toInt
+                    ) throws IOException {
+                        int count = docs.count() - offset;
+                        try (var builder = factory.bytesRefs(count)) {
+                            for (int i = offset; i < docs.count(); i++) {
+                                doc = docs.get(i);
+                                long startOffset = addresses.get(doc);
+                                bytes.length = (int) (addresses.get(doc + 1L) - startOffset);
+                                bytesSlice.readBytes(startOffset, bytes.bytes, 0, bytes.length);
+                                builder.appendBytesRef(bytes);
+                            }
+                            return builder.build();
+                        }
+                    }
                 };
             }
         } else {
@@ -267,7 +309,7 @@ public BytesRef binaryValue() throws IOException {
         }
     }
 
-    private abstract static class DenseBinaryDocValues extends BinaryDocValues {
+    private abstract static class DenseBinaryDocValues extends BinaryDocValues implements BlockLoader.OptionalColumnAtATimeReader {
 
         final int maxDoc;
         int doc = -1;

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

@@ -925,6 +925,28 @@ public AllReader reader(LeafReaderContext context) throws IOException {
         }
     }
 
+    public static class BytesRefsFromBinaryBlockLoader extends DocValuesBlockLoader {
+        private final String fieldName;
+
+        public BytesRefsFromBinaryBlockLoader(String fieldName) {
+            this.fieldName = fieldName;
+        }
+
+        @Override
+        public Builder builder(BlockFactory factory, int expectedCount) {
+            return factory.bytesRefs(expectedCount);
+        }
+
+        @Override
+        public AllReader reader(LeafReaderContext context) throws IOException {
+            BinaryDocValues docValues = context.reader().getBinaryDocValues(fieldName);
+            if (docValues == null) {
+                return new ConstantNullsReader();
+            }
+            return new BytesRefsFromBinary(docValues);
+        }
+    }
+
     abstract static class AbstractBytesRefsFromBinary extends BlockDocValuesReader {
         protected final BinaryDocValues docValues;
 
@@ -1010,6 +1032,17 @@ public BytesRefsFromBinary(BinaryDocValues docValues) {
             super(docValues);
         }
 
+        @Override
+        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFiltered) throws IOException {
+            if (docValues instanceof BlockLoader.OptionalColumnAtATimeReader direct) {
+                BlockLoader.Block block = direct.tryRead(factory, docs, offset, nullsFiltered, null, false);
+                if (block != null) {
+                    return block;
+                }
+            }
+            return super.read(factory, docs, offset, nullsFiltered);
+        }
+
         @Override
         void read(int doc, BytesRefBuilder builder) throws IOException {
             if (false == docValues.advanceExact(doc)) {