Prepare tsdb doc values format for merging optimizations. (#125933)

The change contains the following changes:

- The numDocsWithField field moved from SortedNumericEntry to NumericEntry. Making this statistic always available.
- Store jump table after values in ES87TSDBDocValuesConsumer#writeField(...). Currently it is stored before storing values. This will allow us later to iterate over the SortedNumericDocValues once. When merging, this is expensive as a merge sort on the fly is being executed.

This change will allow all the optimizations that are listed in #125403

server/src/main/java/module-info.java

@@ -443,7 +443,10 @@ module org.elasticsearch.server {
             org.elasticsearch.index.codec.bloomfilter.ES85BloomFilterPostingsFormat,
             org.elasticsearch.index.codec.bloomfilter.ES87BloomFilterPostingsFormat,
             org.elasticsearch.index.codec.postings.ES812PostingsFormat;
-    provides org.apache.lucene.codecs.DocValuesFormat with org.elasticsearch.index.codec.tsdb.ES87TSDBDocValuesFormat;
+    provides org.apache.lucene.codecs.DocValuesFormat
+        with
+            org.elasticsearch.index.codec.tsdb.ES87TSDBDocValuesFormat,
+            org.elasticsearch.index.codec.tsdb.es819.ES819TSDBDocValuesFormat;
     provides org.apache.lucene.codecs.KnnVectorsFormat
         with
             org.elasticsearch.index.codec.vectors.ES813FlatVectorFormat,

server/src/main/java/org/elasticsearch/index/codec/PerFieldFormatSupplier.java

@@ -19,7 +19,7 @@ import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.codec.bloomfilter.ES87BloomFilterPostingsFormat;
 import org.elasticsearch.index.codec.postings.ES812PostingsFormat;
-import org.elasticsearch.index.codec.tsdb.ES87TSDBDocValuesFormat;
+import org.elasticsearch.index.codec.tsdb.es819.ES819TSDBDocValuesFormat;
 import org.elasticsearch.index.mapper.CompletionFieldMapper;
 import org.elasticsearch.index.mapper.IdFieldMapper;
 import org.elasticsearch.index.mapper.Mapper;
@@ -34,7 +34,7 @@ public class PerFieldFormatSupplier {
 
     private static final DocValuesFormat docValuesFormat = new Lucene90DocValuesFormat();
     private static final KnnVectorsFormat knnVectorsFormat = new Lucene99HnswVectorsFormat();
-    private static final ES87TSDBDocValuesFormat tsdbDocValuesFormat = new ES87TSDBDocValuesFormat();
+    private static final ES819TSDBDocValuesFormat tsdbDocValuesFormat = new ES819TSDBDocValuesFormat();
     private static final ES812PostingsFormat es812PostingsFormat = new ES812PostingsFormat();
     private static final PostingsFormat completionPostingsFormat = PostingsFormat.forName("Completion101");
 

server/src/main/java/org/elasticsearch/index/codec/tsdb/ES87TSDBDocValuesFormat.java

@@ -75,7 +75,7 @@ public class ES87TSDBDocValuesFormat extends org.apache.lucene.codecs.DocValuesF
         }
     }
 
-    private final int skipIndexIntervalSize;
+    final int skipIndexIntervalSize;
 
     /** Default constructor. */
     public ES87TSDBDocValuesFormat() {
@@ -93,7 +93,7 @@ public class ES87TSDBDocValuesFormat extends org.apache.lucene.codecs.DocValuesF
 
     @Override
     public DocValuesConsumer fieldsConsumer(SegmentWriteState state) throws IOException {
-        return new ES87TSDBDocValuesConsumer(state, skipIndexIntervalSize, DATA_CODEC, DATA_EXTENSION, META_CODEC, META_EXTENSION);
+        throw new UnsupportedOperationException("writing es87 doc values is no longer supported");
     }
 
     @Override

server/src/main/java/org/elasticsearch/index/codec/tsdb/ES87TSDBDocValuesProducer.java

@@ -49,7 +49,7 @@ import static org.elasticsearch.index.codec.tsdb.ES87TSDBDocValuesFormat.SKIP_IN
 import static org.elasticsearch.index.codec.tsdb.ES87TSDBDocValuesFormat.SKIP_INDEX_MAX_LEVEL;
 import static org.elasticsearch.index.codec.tsdb.ES87TSDBDocValuesFormat.TERMS_DICT_BLOCK_LZ4_SHIFT;
 
-public class ES87TSDBDocValuesProducer extends DocValuesProducer {
+final class ES87TSDBDocValuesProducer extends DocValuesProducer {
     private final IntObjectHashMap<NumericEntry> numerics;
     private final IntObjectHashMap<BinaryEntry> binaries;
     private final IntObjectHashMap<SortedEntry> sorted;

server/src/main/java/org/elasticsearch/index/codec/tsdb/TSDBDocValuesEncoder.java

@@ -175,7 +175,7 @@ public class TSDBDocValuesEncoder {
     /**
      * Encode the given longs using a combination of delta-coding, GCD factorization and bit packing.
      */
-    void encode(long[] in, DataOutput out) throws IOException {
+    public void encode(long[] in, DataOutput out) throws IOException {
         assert in.length == numericBlockSize;
 
         deltaEncode(0, 0, in, out);
@@ -193,7 +193,7 @@ public class TSDBDocValuesEncoder {
      *   <li>3: cycle</li>
      * </ul>
      */
-    void encodeOrdinals(long[] in, DataOutput out, int bitsPerOrd) throws IOException {
+    public void encodeOrdinals(long[] in, DataOutput out, int bitsPerOrd) throws IOException {
         assert in.length == numericBlockSize;
         int numRuns = 1;
         long firstValue = in[0];
@@ -260,7 +260,7 @@ public class TSDBDocValuesEncoder {
         }
     }
 
-    void decodeOrdinals(DataInput in, long[] out, int bitsPerOrd) throws IOException {
+    public void decodeOrdinals(DataInput in, long[] out, int bitsPerOrd) throws IOException {
         assert out.length == numericBlockSize : out.length;
 
         long v1 = in.readVLong();
@@ -294,7 +294,7 @@ public class TSDBDocValuesEncoder {
     }
 
     /** Decode longs that have been encoded with {@link #encode}. */
-    void decode(DataInput in, long[] out) throws IOException {
+    public void decode(DataInput in, long[] out) throws IOException {
         assert out.length == numericBlockSize : out.length;
 
         final int token = in.readVInt();

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

@@ -0,0 +1,790 @@
+/*
+ * 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.codecs.CodecUtil;
+import org.apache.lucene.codecs.DocValuesConsumer;
+import org.apache.lucene.codecs.DocValuesProducer;
+import org.apache.lucene.codecs.lucene90.IndexedDISI;
+import org.apache.lucene.index.BinaryDocValues;
+import org.apache.lucene.index.DocValues;
+import org.apache.lucene.index.DocValuesSkipIndexType;
+import org.apache.lucene.index.EmptyDocValuesProducer;
+import org.apache.lucene.index.FieldInfo;
+import org.apache.lucene.index.IndexFileNames;
+import org.apache.lucene.index.NumericDocValues;
+import org.apache.lucene.index.SegmentWriteState;
+import org.apache.lucene.index.SortedDocValues;
+import org.apache.lucene.index.SortedNumericDocValues;
+import org.apache.lucene.index.SortedSetDocValues;
+import org.apache.lucene.index.TermsEnum;
+import org.apache.lucene.search.DocIdSetIterator;
+import org.apache.lucene.search.SortedSetSelector;
+import org.apache.lucene.store.ByteArrayDataOutput;
+import org.apache.lucene.store.ByteBuffersDataOutput;
+import org.apache.lucene.store.ByteBuffersIndexOutput;
+import org.apache.lucene.store.IndexOutput;
+import org.apache.lucene.util.ArrayUtil;
+import org.apache.lucene.util.BytesRef;
+import org.apache.lucene.util.BytesRefBuilder;
+import org.apache.lucene.util.LongsRef;
+import org.apache.lucene.util.StringHelper;
+import org.apache.lucene.util.compress.LZ4;
+import org.apache.lucene.util.packed.DirectMonotonicWriter;
+import org.apache.lucene.util.packed.PackedInts;
+import org.elasticsearch.core.IOUtils;
+import org.elasticsearch.index.codec.tsdb.TSDBDocValuesEncoder;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.elasticsearch.index.codec.tsdb.es819.ES819TSDBDocValuesFormat.DIRECT_MONOTONIC_BLOCK_SHIFT;
+import static org.elasticsearch.index.codec.tsdb.es819.ES819TSDBDocValuesFormat.SKIP_INDEX_LEVEL_SHIFT;
+import static org.elasticsearch.index.codec.tsdb.es819.ES819TSDBDocValuesFormat.SKIP_INDEX_MAX_LEVEL;
+import static org.elasticsearch.index.codec.tsdb.es819.ES819TSDBDocValuesFormat.SORTED_SET;
+
+final class ES819TSDBDocValuesConsumer extends DocValuesConsumer {
+
+    IndexOutput data, meta;
+    final int maxDoc;
+    private byte[] termsDictBuffer;
+    private final int skipIndexIntervalSize;
+
+    ES819TSDBDocValuesConsumer(
+        SegmentWriteState state,
+        int skipIndexIntervalSize,
+        String dataCodec,
+        String dataExtension,
+        String metaCodec,
+        String metaExtension
+    ) throws IOException {
+        this.termsDictBuffer = new byte[1 << 14];
+        boolean success = false;
+        try {
+            final String dataName = IndexFileNames.segmentFileName(state.segmentInfo.name, state.segmentSuffix, dataExtension);
+            data = state.directory.createOutput(dataName, state.context);
+            CodecUtil.writeIndexHeader(
+                data,
+                dataCodec,
+                ES819TSDBDocValuesFormat.VERSION_CURRENT,
+                state.segmentInfo.getId(),
+                state.segmentSuffix
+            );
+            String metaName = IndexFileNames.segmentFileName(state.segmentInfo.name, state.segmentSuffix, metaExtension);
+            meta = state.directory.createOutput(metaName, state.context);
+            CodecUtil.writeIndexHeader(
+                meta,
+                metaCodec,
+                ES819TSDBDocValuesFormat.VERSION_CURRENT,
+                state.segmentInfo.getId(),
+                state.segmentSuffix
+            );
+            maxDoc = state.segmentInfo.maxDoc();
+            this.skipIndexIntervalSize = skipIndexIntervalSize;
+            success = true;
+        } finally {
+            if (success == false) {
+                IOUtils.closeWhileHandlingException(this);
+            }
+        }
+    }
+
+    @Override
+    public void addNumericField(FieldInfo field, DocValuesProducer valuesProducer) throws IOException {
+        meta.writeInt(field.number);
+        meta.writeByte(ES819TSDBDocValuesFormat.NUMERIC);
+        DocValuesProducer producer = new EmptyDocValuesProducer() {
+            @Override
+            public SortedNumericDocValues getSortedNumeric(FieldInfo field) throws IOException {
+                return DocValues.singleton(valuesProducer.getNumeric(field));
+            }
+        };
+        if (field.docValuesSkipIndexType() != DocValuesSkipIndexType.NONE) {
+            writeSkipIndex(field, producer);
+        }
+
+        writeField(field, producer, -1);
+    }
+
+    private long[] writeField(FieldInfo field, DocValuesProducer valuesProducer, long maxOrd) throws IOException {
+        int numDocsWithValue = 0;
+        long numValues = 0;
+
+        SortedNumericDocValues values = valuesProducer.getSortedNumeric(field);
+        for (int doc = values.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = values.nextDoc()) {
+            numDocsWithValue++;
+            final int count = values.docValueCount();
+            numValues += count;
+        }
+
+        meta.writeLong(numValues);
+        meta.writeInt(numDocsWithValue);
+
+        if (numValues > 0) {
+            // Special case for maxOrd of 1, signal -1 that no blocks will be written
+            meta.writeInt(maxOrd != 1 ? ES819TSDBDocValuesFormat.DIRECT_MONOTONIC_BLOCK_SHIFT : -1);
+            final ByteBuffersDataOutput indexOut = new ByteBuffersDataOutput();
+            final DirectMonotonicWriter indexWriter = DirectMonotonicWriter.getInstance(
+                meta,
+                new ByteBuffersIndexOutput(indexOut, "temp-dv-index", "temp-dv-index"),
+                1L + ((numValues - 1) >>> ES819TSDBDocValuesFormat.NUMERIC_BLOCK_SHIFT),
+                ES819TSDBDocValuesFormat.DIRECT_MONOTONIC_BLOCK_SHIFT
+            );
+
+            final long valuesDataOffset = data.getFilePointer();
+            // Special case for maxOrd of 1, skip writing the blocks
+            if (maxOrd != 1) {
+                final long[] buffer = new long[ES819TSDBDocValuesFormat.NUMERIC_BLOCK_SIZE];
+                int bufferSize = 0;
+                final TSDBDocValuesEncoder encoder = new TSDBDocValuesEncoder(ES819TSDBDocValuesFormat.NUMERIC_BLOCK_SIZE);
+                values = valuesProducer.getSortedNumeric(field);
+                final int bitsPerOrd = maxOrd >= 0 ? PackedInts.bitsRequired(maxOrd - 1) : -1;
+                for (int doc = values.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = values.nextDoc()) {
+                    final int count = values.docValueCount();
+                    for (int i = 0; i < count; ++i) {
+                        buffer[bufferSize++] = values.nextValue();
+                        if (bufferSize == ES819TSDBDocValuesFormat.NUMERIC_BLOCK_SIZE) {
+                            indexWriter.add(data.getFilePointer() - valuesDataOffset);
+                            if (maxOrd >= 0) {
+                                encoder.encodeOrdinals(buffer, data, bitsPerOrd);
+                            } else {
+                                encoder.encode(buffer, data);
+                            }
+                            bufferSize = 0;
+                        }
+                    }
+                }
+                if (bufferSize > 0) {
+                    indexWriter.add(data.getFilePointer() - valuesDataOffset);
+                    // Fill unused slots in the block with zeroes rather than junk
+                    Arrays.fill(buffer, bufferSize, ES819TSDBDocValuesFormat.NUMERIC_BLOCK_SIZE, 0L);
+                    if (maxOrd >= 0) {
+                        encoder.encodeOrdinals(buffer, data, bitsPerOrd);
+                    } else {
+                        encoder.encode(buffer, data);
+                    }
+                }
+            }
+
+            final long valuesDataLength = data.getFilePointer() - valuesDataOffset;
+            if (maxOrd != 1) {
+                // Special case for maxOrd of 1, indexWriter isn't really used, so no need to invoke finish() method.
+                indexWriter.finish();
+            }
+            final long indexDataOffset = data.getFilePointer();
+            data.copyBytes(indexOut.toDataInput(), indexOut.size());
+            meta.writeLong(indexDataOffset);
+            meta.writeLong(data.getFilePointer() - indexDataOffset);
+
+            meta.writeLong(valuesDataOffset);
+            meta.writeLong(valuesDataLength);
+        }
+
+        if (numDocsWithValue == 0) { // meta[-2, 0]: No documents with values
+            meta.writeLong(-2); // docsWithFieldOffset
+            meta.writeLong(0L); // docsWithFieldLength
+            meta.writeShort((short) -1); // jumpTableEntryCount
+            meta.writeByte((byte) -1); // denseRankPower
+        } else if (numDocsWithValue == maxDoc) { // meta[-1, 0]: All documents have values
+            meta.writeLong(-1); // docsWithFieldOffset
+            meta.writeLong(0L); // docsWithFieldLength
+            meta.writeShort((short) -1); // jumpTableEntryCount
+            meta.writeByte((byte) -1); // denseRankPower
+        } else { // meta[data.offset, data.length]: IndexedDISI structure for documents with values
+            long offset = data.getFilePointer();
+            meta.writeLong(offset); // docsWithFieldOffset
+            values = valuesProducer.getSortedNumeric(field);
+            final short jumpTableEntryCount = IndexedDISI.writeBitSet(values, data, IndexedDISI.DEFAULT_DENSE_RANK_POWER);
+            meta.writeLong(data.getFilePointer() - offset); // docsWithFieldLength
+            meta.writeShort(jumpTableEntryCount);
+            meta.writeByte(IndexedDISI.DEFAULT_DENSE_RANK_POWER);
+        }
+
+        return new long[] { numDocsWithValue, numValues };
+    }
+
+    @Override
+    public void addBinaryField(FieldInfo field, DocValuesProducer valuesProducer) throws IOException {
+        meta.writeInt(field.number);
+        meta.writeByte(ES819TSDBDocValuesFormat.BINARY);
+
+        BinaryDocValues values = valuesProducer.getBinary(field);
+        long start = data.getFilePointer();
+        meta.writeLong(start); // dataOffset
+        int numDocsWithField = 0;
+        int minLength = Integer.MAX_VALUE;
+        int maxLength = 0;
+        for (int doc = values.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = values.nextDoc()) {
+            numDocsWithField++;
+            BytesRef v = values.binaryValue();
+            int length = v.length;
+            data.writeBytes(v.bytes, v.offset, v.length);
+            minLength = Math.min(length, minLength);
+            maxLength = Math.max(length, maxLength);
+        }
+        assert numDocsWithField <= maxDoc;
+        meta.writeLong(data.getFilePointer() - start); // dataLength
+
+        if (numDocsWithField == 0) {
+            meta.writeLong(-2); // docsWithFieldOffset
+            meta.writeLong(0L); // docsWithFieldLength
+            meta.writeShort((short) -1); // jumpTableEntryCount
+            meta.writeByte((byte) -1); // denseRankPower
+        } else if (numDocsWithField == maxDoc) {
+            meta.writeLong(-1); // docsWithFieldOffset
+            meta.writeLong(0L); // docsWithFieldLength
+            meta.writeShort((short) -1); // jumpTableEntryCount
+            meta.writeByte((byte) -1); // denseRankPower
+        } else {
+            long offset = data.getFilePointer();
+            meta.writeLong(offset); // docsWithFieldOffset
+            values = valuesProducer.getBinary(field);
+            final short jumpTableEntryCount = IndexedDISI.writeBitSet(values, data, IndexedDISI.DEFAULT_DENSE_RANK_POWER);
+            meta.writeLong(data.getFilePointer() - offset); // docsWithFieldLength
+            meta.writeShort(jumpTableEntryCount);
+            meta.writeByte(IndexedDISI.DEFAULT_DENSE_RANK_POWER);
+        }
+
+        meta.writeInt(numDocsWithField);
+        meta.writeInt(minLength);
+        meta.writeInt(maxLength);
+        if (maxLength > minLength) {
+            start = data.getFilePointer();
+            meta.writeLong(start);
+            meta.writeVInt(ES819TSDBDocValuesFormat.DIRECT_MONOTONIC_BLOCK_SHIFT);
+
+            final DirectMonotonicWriter writer = DirectMonotonicWriter.getInstance(
+                meta,
+                data,
+                numDocsWithField + 1,
+                ES819TSDBDocValuesFormat.DIRECT_MONOTONIC_BLOCK_SHIFT
+            );
+            long addr = 0;
+            writer.add(addr);
+            values = valuesProducer.getBinary(field);
+            for (int doc = values.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = values.nextDoc()) {
+                addr += values.binaryValue().length;
+                writer.add(addr);
+            }
+            writer.finish();
+            meta.writeLong(data.getFilePointer() - start);
+        }
+    }
+
+    @Override
+    public void addSortedField(FieldInfo field, DocValuesProducer valuesProducer) throws IOException {
+        meta.writeInt(field.number);
+        meta.writeByte(ES819TSDBDocValuesFormat.SORTED);
+        doAddSortedField(field, valuesProducer, false);
+    }
+
+    private void doAddSortedField(FieldInfo field, DocValuesProducer valuesProducer, boolean addTypeByte) throws IOException {
+        DocValuesProducer producer = new EmptyDocValuesProducer() {
+            @Override
+            public SortedNumericDocValues getSortedNumeric(FieldInfo field) throws IOException {
+                SortedDocValues sorted = valuesProducer.getSorted(field);
+                NumericDocValues sortedOrds = new NumericDocValues() {
+                    @Override
+                    public long longValue() throws IOException {
+                        return sorted.ordValue();
+                    }
+
+                    @Override
+                    public boolean advanceExact(int target) throws IOException {
+                        return sorted.advanceExact(target);
+                    }
+
+                    @Override
+                    public int docID() {
+                        return sorted.docID();
+                    }
+
+                    @Override
+                    public int nextDoc() throws IOException {
+                        return sorted.nextDoc();
+                    }
+
+                    @Override
+                    public int advance(int target) throws IOException {
+                        return sorted.advance(target);
+                    }
+
+                    @Override
+                    public long cost() {
+                        return sorted.cost();
+                    }
+                };
+                return DocValues.singleton(sortedOrds);
+            }
+        };
+        if (field.docValuesSkipIndexType() != DocValuesSkipIndexType.NONE) {
+            writeSkipIndex(field, producer);
+        }
+        if (addTypeByte) {
+            meta.writeByte((byte) 0); // multiValued (0 = singleValued)
+        }
+        SortedDocValues sorted = valuesProducer.getSorted(field);
+        int maxOrd = sorted.getValueCount();
+        writeField(field, producer, maxOrd);
+        addTermsDict(DocValues.singleton(valuesProducer.getSorted(field)));
+    }
+
+    private void addTermsDict(SortedSetDocValues values) throws IOException {
+        final long size = values.getValueCount();
+        meta.writeVLong(size);
+
+        int blockMask = ES819TSDBDocValuesFormat.TERMS_DICT_BLOCK_LZ4_MASK;
+        int shift = ES819TSDBDocValuesFormat.TERMS_DICT_BLOCK_LZ4_SHIFT;
+
+        meta.writeInt(DIRECT_MONOTONIC_BLOCK_SHIFT);
+        ByteBuffersDataOutput addressBuffer = new ByteBuffersDataOutput();
+        ByteBuffersIndexOutput addressOutput = new ByteBuffersIndexOutput(addressBuffer, "temp", "temp");
+        long numBlocks = (size + blockMask) >>> shift;
+        DirectMonotonicWriter writer = DirectMonotonicWriter.getInstance(meta, addressOutput, numBlocks, DIRECT_MONOTONIC_BLOCK_SHIFT);
+
+        BytesRefBuilder previous = new BytesRefBuilder();
+        long ord = 0;
+        long start = data.getFilePointer();
+        int maxLength = 0, maxBlockLength = 0;
+        TermsEnum iterator = values.termsEnum();
+
+        LZ4.FastCompressionHashTable ht = new LZ4.FastCompressionHashTable();
+        ByteArrayDataOutput bufferedOutput = new ByteArrayDataOutput(termsDictBuffer);
+        int dictLength = 0;
+
+        for (BytesRef term = iterator.next(); term != null; term = iterator.next()) {
+            if ((ord & blockMask) == 0) {
+                if (ord != 0) {
+                    // flush the previous block
+                    final int uncompressedLength = compressAndGetTermsDictBlockLength(bufferedOutput, dictLength, ht);
+                    maxBlockLength = Math.max(maxBlockLength, uncompressedLength);
+                    bufferedOutput.reset(termsDictBuffer);
+                }
+
+                writer.add(data.getFilePointer() - start);
+                // Write the first term both to the index output, and to the buffer where we'll use it as a
+                // dictionary for compression
+                data.writeVInt(term.length);
+                data.writeBytes(term.bytes, term.offset, term.length);
+                bufferedOutput = maybeGrowBuffer(bufferedOutput, term.length);
+                bufferedOutput.writeBytes(term.bytes, term.offset, term.length);
+                dictLength = term.length;
+            } else {
+                final int prefixLength = StringHelper.bytesDifference(previous.get(), term);
+                final int suffixLength = term.length - prefixLength;
+                assert suffixLength > 0; // terms are unique
+                // Will write (suffixLength + 1 byte + 2 vint) bytes. Grow the buffer in need.
+                bufferedOutput = maybeGrowBuffer(bufferedOutput, suffixLength + 11);
+                bufferedOutput.writeByte((byte) (Math.min(prefixLength, 15) | (Math.min(15, suffixLength - 1) << 4)));
+                if (prefixLength >= 15) {
+                    bufferedOutput.writeVInt(prefixLength - 15);
+                }
+                if (suffixLength >= 16) {
+                    bufferedOutput.writeVInt(suffixLength - 16);
+                }
+                bufferedOutput.writeBytes(term.bytes, term.offset + prefixLength, suffixLength);
+            }
+            maxLength = Math.max(maxLength, term.length);
+            previous.copyBytes(term);
+            ++ord;
+        }
+        // Compress and write out the last block
+        if (bufferedOutput.getPosition() > dictLength) {
+            final int uncompressedLength = compressAndGetTermsDictBlockLength(bufferedOutput, dictLength, ht);
+            maxBlockLength = Math.max(maxBlockLength, uncompressedLength);
+        }
+
+        writer.finish();
+        meta.writeInt(maxLength);
+        // Write one more int for storing max block length.
+        meta.writeInt(maxBlockLength);
+        meta.writeLong(start);
+        meta.writeLong(data.getFilePointer() - start);
+        start = data.getFilePointer();
+        addressBuffer.copyTo(data);
+        meta.writeLong(start);
+        meta.writeLong(data.getFilePointer() - start);
+
+        // Now write the reverse terms index
+        writeTermsIndex(values);
+    }
+
+    private int compressAndGetTermsDictBlockLength(ByteArrayDataOutput bufferedOutput, int dictLength, LZ4.FastCompressionHashTable ht)
+        throws IOException {
+        int uncompressedLength = bufferedOutput.getPosition() - dictLength;
+        data.writeVInt(uncompressedLength);
+        LZ4.compressWithDictionary(termsDictBuffer, 0, dictLength, uncompressedLength, data, ht);
+        return uncompressedLength;
+    }
+
+    private ByteArrayDataOutput maybeGrowBuffer(ByteArrayDataOutput bufferedOutput, int termLength) {
+        int pos = bufferedOutput.getPosition(), originalLength = termsDictBuffer.length;
+        if (pos + termLength >= originalLength - 1) {
+            termsDictBuffer = ArrayUtil.grow(termsDictBuffer, originalLength + termLength);
+            bufferedOutput = new ByteArrayDataOutput(termsDictBuffer, pos, termsDictBuffer.length - pos);
+        }
+        return bufferedOutput;
+    }
+
+    private void writeTermsIndex(SortedSetDocValues values) throws IOException {
+        final long size = values.getValueCount();
+        meta.writeInt(ES819TSDBDocValuesFormat.TERMS_DICT_REVERSE_INDEX_SHIFT);
+        long start = data.getFilePointer();
+
+        long numBlocks = 1L + ((size + ES819TSDBDocValuesFormat.TERMS_DICT_REVERSE_INDEX_MASK)
+            >>> ES819TSDBDocValuesFormat.TERMS_DICT_REVERSE_INDEX_SHIFT);
+        ByteBuffersDataOutput addressBuffer = new ByteBuffersDataOutput();
+        DirectMonotonicWriter writer;
+        try (ByteBuffersIndexOutput addressOutput = new ByteBuffersIndexOutput(addressBuffer, "temp", "temp")) {
+            writer = DirectMonotonicWriter.getInstance(meta, addressOutput, numBlocks, DIRECT_MONOTONIC_BLOCK_SHIFT);
+            TermsEnum iterator = values.termsEnum();
+            BytesRefBuilder previous = new BytesRefBuilder();
+            long offset = 0;
+            long ord = 0;
+            for (BytesRef term = iterator.next(); term != null; term = iterator.next()) {
+                if ((ord & ES819TSDBDocValuesFormat.TERMS_DICT_REVERSE_INDEX_MASK) == 0) {
+                    writer.add(offset);
+                    final int sortKeyLength;
+                    if (ord == 0) {
+                        // no previous term: no bytes to write
+                        sortKeyLength = 0;
+                    } else {
+                        sortKeyLength = StringHelper.sortKeyLength(previous.get(), term);
+                    }
+                    offset += sortKeyLength;
+                    data.writeBytes(term.bytes, term.offset, sortKeyLength);
+                } else if ((ord
+                    & ES819TSDBDocValuesFormat.TERMS_DICT_REVERSE_INDEX_MASK) == ES819TSDBDocValuesFormat.TERMS_DICT_REVERSE_INDEX_MASK) {
+                        previous.copyBytes(term);
+                    }
+                ++ord;
+            }
+            writer.add(offset);
+            writer.finish();
+            meta.writeLong(start);
+            meta.writeLong(data.getFilePointer() - start);
+            start = data.getFilePointer();
+            addressBuffer.copyTo(data);
+            meta.writeLong(start);
+            meta.writeLong(data.getFilePointer() - start);
+        }
+    }
+
+    @Override
+    public void addSortedNumericField(FieldInfo field, DocValuesProducer valuesProducer) throws IOException {
+        meta.writeInt(field.number);
+        meta.writeByte(ES819TSDBDocValuesFormat.SORTED_NUMERIC);
+        writeSortedNumericField(field, valuesProducer, -1);
+    }
+
+    private void writeSortedNumericField(FieldInfo field, DocValuesProducer valuesProducer, long maxOrd) throws IOException {
+        if (field.docValuesSkipIndexType() != DocValuesSkipIndexType.NONE) {
+            writeSkipIndex(field, valuesProducer);
+        }
+        if (maxOrd > -1) {
+            meta.writeByte((byte) 1); // multiValued (1 = multiValued)
+        }
+        long[] stats = writeField(field, valuesProducer, maxOrd);
+        int numDocsWithField = Math.toIntExact(stats[0]);
+        long numValues = stats[1];
+        assert numValues >= numDocsWithField;
+
+        if (numValues > numDocsWithField) {
+            long start = data.getFilePointer();
+            meta.writeLong(start);
+            meta.writeVInt(ES819TSDBDocValuesFormat.DIRECT_MONOTONIC_BLOCK_SHIFT);
+
+            final DirectMonotonicWriter addressesWriter = DirectMonotonicWriter.getInstance(
+                meta,
+                data,
+                numDocsWithField + 1L,
+                ES819TSDBDocValuesFormat.DIRECT_MONOTONIC_BLOCK_SHIFT
+            );
+            long addr = 0;
+            addressesWriter.add(addr);
+            SortedNumericDocValues values = valuesProducer.getSortedNumeric(field);
+            for (int doc = values.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = values.nextDoc()) {
+                addr += values.docValueCount();
+                addressesWriter.add(addr);
+            }
+            addressesWriter.finish();
+            meta.writeLong(data.getFilePointer() - start);
+        }
+    }
+
+    private static boolean isSingleValued(SortedSetDocValues values) throws IOException {
+        if (DocValues.unwrapSingleton(values) != null) {
+            return true;
+        }
+
+        assert values.docID() == -1;
+        for (int doc = values.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = values.nextDoc()) {
+            int docValueCount = values.docValueCount();
+            assert docValueCount > 0;
+            if (docValueCount > 1) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    @Override
+    public void addSortedSetField(FieldInfo field, DocValuesProducer valuesProducer) throws IOException {
+        meta.writeInt(field.number);
+        meta.writeByte(SORTED_SET);
+
+        if (isSingleValued(valuesProducer.getSortedSet(field))) {
+            doAddSortedField(field, new EmptyDocValuesProducer() {
+                @Override
+                public SortedDocValues getSorted(FieldInfo field) throws IOException {
+                    return SortedSetSelector.wrap(valuesProducer.getSortedSet(field), SortedSetSelector.Type.MIN);
+                }
+            }, true);
+            return;
+        }
+
+        SortedSetDocValues values = valuesProducer.getSortedSet(field);
+        long maxOrd = values.getValueCount();
+        writeSortedNumericField(field, new EmptyDocValuesProducer() {
+            @Override
+            public SortedNumericDocValues getSortedNumeric(FieldInfo field) throws IOException {
+                SortedSetDocValues values = valuesProducer.getSortedSet(field);
+                return new SortedNumericDocValues() {
+
+                    long[] ords = LongsRef.EMPTY_LONGS;
+                    int i, docValueCount;
+
+                    @Override
+                    public long nextValue() {
+                        return ords[i++];
+                    }
+
+                    @Override
+                    public int docValueCount() {
+                        return docValueCount;
+                    }
+
+                    @Override
+                    public boolean advanceExact(int target) {
+                        throw new UnsupportedOperationException();
+                    }
+
+                    @Override
+                    public int docID() {
+                        return values.docID();
+                    }
+
+                    @Override
+                    public int nextDoc() throws IOException {
+                        int doc = values.nextDoc();
+                        if (doc != NO_MORE_DOCS) {
+                            docValueCount = values.docValueCount();
+                            ords = ArrayUtil.grow(ords, docValueCount);
+                            for (int j = 0; j < docValueCount; j++) {
+                                ords[j] = values.nextOrd();
+                            }
+                            i = 0;
+                        }
+                        return doc;
+                    }
+
+                    @Override
+                    public int advance(int target) throws IOException {
+                        throw new UnsupportedOperationException();
+                    }
+
+                    @Override
+                    public long cost() {
+                        return values.cost();
+                    }
+                };
+            }
+        }, maxOrd);
+
+        addTermsDict(valuesProducer.getSortedSet(field));
+    }
+
+    @Override
+    public void close() throws IOException {
+        boolean success = false;
+        try {
+            if (meta != null) {
+                meta.writeInt(-1); // write EOF marker
+                CodecUtil.writeFooter(meta); // write checksum
+            }
+            if (data != null) {
+                CodecUtil.writeFooter(data); // write checksum
+            }
+            success = true;
+        } finally {
+            if (success) {
+                IOUtils.close(data, meta);
+            } else {
+                IOUtils.closeWhileHandlingException(data, meta);
+            }
+            meta = data = null;
+        }
+    }
+
+    private static class SkipAccumulator {
+        int minDocID;
+        int maxDocID;
+        int docCount;
+        long minValue;
+        long maxValue;
+
+        SkipAccumulator(int docID) {
+            minDocID = docID;
+            minValue = Long.MAX_VALUE;
+            maxValue = Long.MIN_VALUE;
+            docCount = 0;
+        }
+
+        boolean isDone(int skipIndexIntervalSize, int valueCount, long nextValue, int nextDoc) {
+            if (docCount < skipIndexIntervalSize) {
+                return false;
+            }
+            // Once we reach the interval size, we will keep accepting documents if
+            // - next doc value is not a multi-value
+            // - current accumulator only contains a single value and next value is the same value
+            // - the accumulator is dense and the next doc keeps the density (no gaps)
+            return valueCount > 1 || minValue != maxValue || minValue != nextValue || docCount != nextDoc - minDocID;
+        }
+
+        void accumulate(long value) {
+            minValue = Math.min(minValue, value);
+            maxValue = Math.max(maxValue, value);
+        }
+
+        void accumulate(SkipAccumulator other) {
+            assert minDocID <= other.minDocID && maxDocID < other.maxDocID;
+            maxDocID = other.maxDocID;
+            minValue = Math.min(minValue, other.minValue);
+            maxValue = Math.max(maxValue, other.maxValue);
+            docCount += other.docCount;
+        }
+
+        void nextDoc(int docID) {
+            maxDocID = docID;
+            ++docCount;
+        }
+
+        public static SkipAccumulator merge(List<SkipAccumulator> list, int index, int length) {
+            SkipAccumulator acc = new SkipAccumulator(list.get(index).minDocID);
+            for (int i = 0; i < length; i++) {
+                acc.accumulate(list.get(index + i));
+            }
+            return acc;
+        }
+    }
+
+    private void writeSkipIndex(FieldInfo field, DocValuesProducer valuesProducer) throws IOException {
+        assert field.docValuesSkipIndexType() != DocValuesSkipIndexType.NONE;
+        final long start = data.getFilePointer();
+        final SortedNumericDocValues values = valuesProducer.getSortedNumeric(field);
+        long globalMaxValue = Long.MIN_VALUE;
+        long globalMinValue = Long.MAX_VALUE;
+        int globalDocCount = 0;
+        int maxDocId = -1;
+        final List<SkipAccumulator> accumulators = new ArrayList<>();
+        SkipAccumulator accumulator = null;
+        final int maxAccumulators = 1 << (SKIP_INDEX_LEVEL_SHIFT * (SKIP_INDEX_MAX_LEVEL - 1));
+        for (int doc = values.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = values.nextDoc()) {
+            final long firstValue = values.nextValue();
+            if (accumulator != null && accumulator.isDone(skipIndexIntervalSize, values.docValueCount(), firstValue, doc)) {
+                globalMaxValue = Math.max(globalMaxValue, accumulator.maxValue);
+                globalMinValue = Math.min(globalMinValue, accumulator.minValue);
+                globalDocCount += accumulator.docCount;
+                maxDocId = accumulator.maxDocID;
+                accumulator = null;
+                if (accumulators.size() == maxAccumulators) {
+                    writeLevels(accumulators);
+                    accumulators.clear();
+                }
+            }
+            if (accumulator == null) {
+                accumulator = new SkipAccumulator(doc);
+                accumulators.add(accumulator);
+            }
+            accumulator.nextDoc(doc);
+            accumulator.accumulate(firstValue);
+            for (int i = 1, end = values.docValueCount(); i < end; ++i) {
+                accumulator.accumulate(values.nextValue());
+            }
+        }
+
+        if (accumulators.isEmpty() == false) {
+            globalMaxValue = Math.max(globalMaxValue, accumulator.maxValue);
+            globalMinValue = Math.min(globalMinValue, accumulator.minValue);
+            globalDocCount += accumulator.docCount;
+            maxDocId = accumulator.maxDocID;
+            writeLevels(accumulators);
+        }
+        meta.writeLong(start); // record the start in meta
+        meta.writeLong(data.getFilePointer() - start); // record the length
+        assert globalDocCount == 0 || globalMaxValue >= globalMinValue;
+        meta.writeLong(globalMaxValue);
+        meta.writeLong(globalMinValue);
+        assert globalDocCount <= maxDocId + 1;
+        meta.writeInt(globalDocCount);
+        meta.writeInt(maxDocId);
+    }
+
+    private void writeLevels(List<SkipAccumulator> accumulators) throws IOException {
+        final List<List<SkipAccumulator>> accumulatorsLevels = new ArrayList<>(SKIP_INDEX_MAX_LEVEL);
+        accumulatorsLevels.add(accumulators);
+        for (int i = 0; i < SKIP_INDEX_MAX_LEVEL - 1; i++) {
+            accumulatorsLevels.add(buildLevel(accumulatorsLevels.get(i)));
+        }
+        int totalAccumulators = accumulators.size();
+        for (int index = 0; index < totalAccumulators; index++) {
+            // compute how many levels we need to write for the current accumulator
+            final int levels = getLevels(index, totalAccumulators);
+            // write the number of levels
+            data.writeByte((byte) levels);
+            // write intervals in reverse order. This is done so we don't
+            // need to read all of them in case of slipping
+            for (int level = levels - 1; level >= 0; level--) {
+                final SkipAccumulator accumulator = accumulatorsLevels.get(level).get(index >> (SKIP_INDEX_LEVEL_SHIFT * level));
+                data.writeInt(accumulator.maxDocID);
+                data.writeInt(accumulator.minDocID);
+                data.writeLong(accumulator.maxValue);
+                data.writeLong(accumulator.minValue);
+                data.writeInt(accumulator.docCount);
+            }
+        }
+    }
+
+    private static List<SkipAccumulator> buildLevel(List<SkipAccumulator> accumulators) {
+        final int levelSize = 1 << SKIP_INDEX_LEVEL_SHIFT;
+        final List<SkipAccumulator> collector = new ArrayList<>();
+        for (int i = 0; i < accumulators.size() - levelSize + 1; i += levelSize) {
+            collector.add(SkipAccumulator.merge(accumulators, i, levelSize));
+        }
+        return collector;
+    }
+
+    private static int getLevels(int index, int size) {
+        if (Integer.numberOfTrailingZeros(index) >= SKIP_INDEX_LEVEL_SHIFT) {
+            // TODO: can we do it in constant time rather than linearly with SKIP_INDEX_MAX_LEVEL?
+            final int left = size - index;
+            for (int level = SKIP_INDEX_MAX_LEVEL - 1; level > 0; level--) {
+                final int numberIntervals = 1 << (SKIP_INDEX_LEVEL_SHIFT * level);
+                if (left >= numberIntervals && index % numberIntervals == 0) {
+                    return level + 1;
+                }
+            }
+        }
+        return 1;
+    }
+
+}

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

@@ -0,0 +1,116 @@
+/*
+ * 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.codecs.DocValuesConsumer;
+import org.apache.lucene.codecs.DocValuesProducer;
+import org.apache.lucene.index.SegmentReadState;
+import org.apache.lucene.index.SegmentWriteState;
+
+import java.io.IOException;
+
+/**
+ * Evolved from {@link org.elasticsearch.index.codec.tsdb.ES87TSDBDocValuesFormat} and has the following changes:
+ * <ul>
+ *     <li>Moved numDocsWithField metadata statistic from SortedNumericEntry to NumericEntry. This allows for always summing
+ *     numDocsWithField during segment merging, otherwise numDocsWithField needs to be computed for each segment merge per field.</li>
+ *     <li>Moved docsWithFieldOffset, docsWithFieldLength, jumpTableEntryCount, denseRankPower metadata properties in the format to be
+ *     after values metadata. So that the jump table can be stored after the values, which allows for iterating once over the merged
+ *     view of all values. If index sorting is active merging a doc value field requires a merge sort which can be very cpu intensive.
+ *     The previous format always has to merge sort a doc values field multiple times, so doing the merge sort just once saves on
+ *     cpu resources.</li>
+ * </ul>
+ */
+public class ES819TSDBDocValuesFormat extends org.apache.lucene.codecs.DocValuesFormat {
+
+    static final int NUMERIC_BLOCK_SHIFT = 7;
+    public static final int NUMERIC_BLOCK_SIZE = 1 << NUMERIC_BLOCK_SHIFT;
+    static final int NUMERIC_BLOCK_MASK = NUMERIC_BLOCK_SIZE - 1;
+    static final int DIRECT_MONOTONIC_BLOCK_SHIFT = 16;
+    static final String CODEC_NAME = "ES819TSDB";
+    static final String DATA_CODEC = "ES819TSDBDocValuesData";
+    static final String DATA_EXTENSION = "dvd";
+    static final String META_CODEC = "ES819TSDBDocValuesMetadata";
+    static final String META_EXTENSION = "dvm";
+    static final byte NUMERIC = 0;
+    static final byte BINARY = 1;
+    static final byte SORTED = 2;
+    static final byte SORTED_SET = 3;
+    static final byte SORTED_NUMERIC = 4;
+
+    static final int VERSION_START = 0;
+    static final int VERSION_CURRENT = VERSION_START;
+
+    static final int TERMS_DICT_BLOCK_LZ4_SHIFT = 6;
+    static final int TERMS_DICT_BLOCK_LZ4_SIZE = 1 << TERMS_DICT_BLOCK_LZ4_SHIFT;
+    static final int TERMS_DICT_BLOCK_LZ4_MASK = TERMS_DICT_BLOCK_LZ4_SIZE - 1;
+
+    static final int TERMS_DICT_REVERSE_INDEX_SHIFT = 10;
+    static final int TERMS_DICT_REVERSE_INDEX_SIZE = 1 << TERMS_DICT_REVERSE_INDEX_SHIFT;
+    static final int TERMS_DICT_REVERSE_INDEX_MASK = TERMS_DICT_REVERSE_INDEX_SIZE - 1;
+
+    // number of documents in an interval
+    private static final int DEFAULT_SKIP_INDEX_INTERVAL_SIZE = 4096;
+    // bytes on an interval:
+    // * 1 byte : number of levels
+    // * 16 bytes: min / max value,
+    // * 8 bytes: min / max docID
+    // * 4 bytes: number of documents
+    private static final long SKIP_INDEX_INTERVAL_BYTES = 29L;
+    // number of intervals represented as a shift to create a new level, this is 1 << 3 == 8
+    // intervals.
+    static final int SKIP_INDEX_LEVEL_SHIFT = 3;
+    // max number of levels
+    // Increasing this number, it increases how much heap we need at index time.
+    // we currently need (1 * 8 * 8 * 8) = 512 accumulators on heap
+    static final int SKIP_INDEX_MAX_LEVEL = 4;
+    // number of bytes to skip when skipping a level. It does not take into account the
+    // current interval that is being read.
+    static final long[] SKIP_INDEX_JUMP_LENGTH_PER_LEVEL = new long[SKIP_INDEX_MAX_LEVEL];
+
+    static {
+        // Size of the interval minus read bytes (1 byte for level and 4 bytes for maxDocID)
+        SKIP_INDEX_JUMP_LENGTH_PER_LEVEL[0] = SKIP_INDEX_INTERVAL_BYTES - 5L;
+        for (int level = 1; level < SKIP_INDEX_MAX_LEVEL; level++) {
+            // jump from previous level
+            SKIP_INDEX_JUMP_LENGTH_PER_LEVEL[level] = SKIP_INDEX_JUMP_LENGTH_PER_LEVEL[level - 1];
+            // nodes added by new level
+            SKIP_INDEX_JUMP_LENGTH_PER_LEVEL[level] += (1 << (level * SKIP_INDEX_LEVEL_SHIFT)) * SKIP_INDEX_INTERVAL_BYTES;
+            // remove the byte levels added in the previous level
+            SKIP_INDEX_JUMP_LENGTH_PER_LEVEL[level] -= (1 << ((level - 1) * SKIP_INDEX_LEVEL_SHIFT));
+        }
+    }
+
+    final int skipIndexIntervalSize;
+
+    /** Default constructor. */
+    public ES819TSDBDocValuesFormat() {
+        this(DEFAULT_SKIP_INDEX_INTERVAL_SIZE);
+    }
+
+    /** Doc values fields format with specified skipIndexIntervalSize. */
+    public ES819TSDBDocValuesFormat(int skipIndexIntervalSize) {
+        super(CODEC_NAME);
+        if (skipIndexIntervalSize < 2) {
+            throw new IllegalArgumentException("skipIndexIntervalSize must be > 1, got [" + skipIndexIntervalSize + "]");
+        }
+        this.skipIndexIntervalSize = skipIndexIntervalSize;
+    }
+
+    @Override
+    public DocValuesConsumer fieldsConsumer(SegmentWriteState state) throws IOException {
+        return new ES819TSDBDocValuesConsumer(state, skipIndexIntervalSize, DATA_CODEC, DATA_EXTENSION, META_CODEC, META_EXTENSION);
+    }
+
+    @Override
+    public DocValuesProducer fieldsProducer(SegmentReadState state) throws IOException {
+        return new ES819TSDBDocValuesProducer(state, DATA_CODEC, DATA_EXTENSION, META_CODEC, META_EXTENSION);
+    }
+}

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

@@ -0,0 +1,1498 @@
+/*
+ * 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.codecs.CodecUtil;
+import org.apache.lucene.codecs.DocValuesProducer;
+import org.apache.lucene.codecs.lucene90.IndexedDISI;
+import org.apache.lucene.index.BaseTermsEnum;
+import org.apache.lucene.index.BinaryDocValues;
+import org.apache.lucene.index.CorruptIndexException;
+import org.apache.lucene.index.DocValues;
+import org.apache.lucene.index.DocValuesSkipIndexType;
+import org.apache.lucene.index.DocValuesSkipper;
+import org.apache.lucene.index.FieldInfo;
+import org.apache.lucene.index.FieldInfos;
+import org.apache.lucene.index.ImpactsEnum;
+import org.apache.lucene.index.IndexFileNames;
+import org.apache.lucene.index.NumericDocValues;
+import org.apache.lucene.index.PostingsEnum;
+import org.apache.lucene.index.SegmentReadState;
+import org.apache.lucene.index.SortedDocValues;
+import org.apache.lucene.index.SortedNumericDocValues;
+import org.apache.lucene.index.SortedSetDocValues;
+import org.apache.lucene.index.TermsEnum;
+import org.apache.lucene.internal.hppc.IntObjectHashMap;
+import org.apache.lucene.search.DocIdSetIterator;
+import org.apache.lucene.store.ByteArrayDataInput;
+import org.apache.lucene.store.ChecksumIndexInput;
+import org.apache.lucene.store.DataInput;
+import org.apache.lucene.store.IndexInput;
+import org.apache.lucene.store.RandomAccessInput;
+import org.apache.lucene.util.BytesRef;
+import org.apache.lucene.util.LongValues;
+import org.apache.lucene.util.compress.LZ4;
+import org.apache.lucene.util.packed.DirectMonotonicReader;
+import org.apache.lucene.util.packed.PackedInts;
+import org.elasticsearch.core.IOUtils;
+import org.elasticsearch.index.codec.tsdb.TSDBDocValuesEncoder;
+
+import java.io.IOException;
+
+import static org.elasticsearch.index.codec.tsdb.es819.ES819TSDBDocValuesFormat.SKIP_INDEX_JUMP_LENGTH_PER_LEVEL;
+import static org.elasticsearch.index.codec.tsdb.es819.ES819TSDBDocValuesFormat.SKIP_INDEX_MAX_LEVEL;
+import static org.elasticsearch.index.codec.tsdb.es819.ES819TSDBDocValuesFormat.TERMS_DICT_BLOCK_LZ4_SHIFT;
+
+final class ES819TSDBDocValuesProducer extends DocValuesProducer {
+    private final IntObjectHashMap<NumericEntry> numerics;
+    private final IntObjectHashMap<BinaryEntry> binaries;
+    private final IntObjectHashMap<SortedEntry> sorted;
+    private final IntObjectHashMap<SortedSetEntry> sortedSets;
+    private final IntObjectHashMap<SortedNumericEntry> sortedNumerics;
+    private final IntObjectHashMap<DocValuesSkipperEntry> skippers;
+    private final IndexInput data;
+    private final int maxDoc;
+    final int version;
+    private final boolean merging;
+
+    ES819TSDBDocValuesProducer(SegmentReadState state, String dataCodec, String dataExtension, String metaCodec, String metaExtension)
+        throws IOException {
+        this.numerics = new IntObjectHashMap<>();
+        this.binaries = new IntObjectHashMap<>();
+        this.sorted = new IntObjectHashMap<>();
+        this.sortedSets = new IntObjectHashMap<>();
+        this.sortedNumerics = new IntObjectHashMap<>();
+        this.skippers = new IntObjectHashMap<>();
+        this.maxDoc = state.segmentInfo.maxDoc();
+        this.merging = false;
+
+        // read in the entries from the metadata file.
+        int version = -1;
+        String metaName = IndexFileNames.segmentFileName(state.segmentInfo.name, state.segmentSuffix, metaExtension);
+        try (ChecksumIndexInput in = state.directory.openChecksumInput(metaName)) {
+            Throwable priorE = null;
+
+            try {
+                version = CodecUtil.checkIndexHeader(
+                    in,
+                    metaCodec,
+                    ES819TSDBDocValuesFormat.VERSION_START,
+                    ES819TSDBDocValuesFormat.VERSION_CURRENT,
+                    state.segmentInfo.getId(),
+                    state.segmentSuffix
+                );
+
+                readFields(in, state.fieldInfos);
+
+            } catch (Throwable exception) {
+                priorE = exception;
+            } finally {
+                CodecUtil.checkFooter(in, priorE);
+            }
+        }
+
+        String dataName = IndexFileNames.segmentFileName(state.segmentInfo.name, state.segmentSuffix, dataExtension);
+        this.data = state.directory.openInput(dataName, state.context);
+        boolean success = false;
+        try {
+            final int version2 = CodecUtil.checkIndexHeader(
+                data,
+                dataCodec,
+                ES819TSDBDocValuesFormat.VERSION_START,
+                ES819TSDBDocValuesFormat.VERSION_CURRENT,
+                state.segmentInfo.getId(),
+                state.segmentSuffix
+            );
+            if (version != version2) {
+                throw new CorruptIndexException("Format versions mismatch: meta=" + version + ", data=" + version2, data);
+            }
+
+            // NOTE: data file is too costly to verify checksum against all the bytes on open,
+            // but for now we at least verify proper structure of the checksum footer: which looks
+            // for FOOTER_MAGIC + algorithmID. This is cheap and can detect some forms of corruption
+            // such as file truncation.
+            CodecUtil.retrieveChecksum(data);
+
+            success = true;
+            this.version = version;
+        } finally {
+            if (success == false) {
+                IOUtils.closeWhileHandlingException(this.data);
+            }
+        }
+    }
+
+    private ES819TSDBDocValuesProducer(
+        IntObjectHashMap<NumericEntry> numerics,
+        IntObjectHashMap<BinaryEntry> binaries,
+        IntObjectHashMap<SortedEntry> sorted,
+        IntObjectHashMap<SortedSetEntry> sortedSets,
+        IntObjectHashMap<SortedNumericEntry> sortedNumerics,
+        IntObjectHashMap<DocValuesSkipperEntry> skippers,
+        IndexInput data,
+        int maxDoc,
+        int version,
+        boolean merging
+    ) {
+        this.numerics = numerics;
+        this.binaries = binaries;
+        this.sorted = sorted;
+        this.sortedSets = sortedSets;
+        this.sortedNumerics = sortedNumerics;
+        this.skippers = skippers;
+        this.data = data.clone();
+        this.maxDoc = maxDoc;
+        this.version = version;
+        this.merging = merging;
+    }
+
+    @Override
+    public DocValuesProducer getMergeInstance() {
+        return new ES819TSDBDocValuesProducer(
+            numerics,
+            binaries,
+            sorted,
+            sortedSets,
+            sortedNumerics,
+            skippers,
+            data,
+            maxDoc,
+            version,
+            true
+        );
+    }
+
+    @Override
+    public NumericDocValues getNumeric(FieldInfo field) throws IOException {
+        NumericEntry entry = numerics.get(field.number);
+        return getNumeric(entry, -1);
+    }
+
+    @Override
+    public BinaryDocValues getBinary(FieldInfo field) throws IOException {
+        BinaryEntry entry = binaries.get(field.number);
+        if (entry.docsWithFieldOffset == -2) {
+            return DocValues.emptyBinary();
+        }
+
+        final RandomAccessInput bytesSlice = data.randomAccessSlice(entry.dataOffset, entry.dataLength);
+
+        if (entry.docsWithFieldOffset == -1) {
+            // dense
+            if (entry.minLength == entry.maxLength) {
+                // fixed length
+                final int length = entry.maxLength;
+                return new DenseBinaryDocValues(maxDoc) {
+                    final BytesRef bytes = new BytesRef(new byte[length], 0, length);
+
+                    @Override
+                    public BytesRef binaryValue() throws IOException {
+                        bytesSlice.readBytes((long) doc * length, bytes.bytes, 0, length);
+                        return bytes;
+                    }
+                };
+            } else {
+                // variable length
+                final RandomAccessInput addressesData = this.data.randomAccessSlice(entry.addressesOffset, entry.addressesLength);
+                final LongValues addresses = DirectMonotonicReader.getInstance(entry.addressesMeta, addressesData, merging);
+                return new DenseBinaryDocValues(maxDoc) {
+                    final BytesRef bytes = new BytesRef(new byte[entry.maxLength], 0, entry.maxLength);
+
+                    @Override
+                    public BytesRef binaryValue() throws IOException {
+                        long startOffset = addresses.get(doc);
+                        bytes.length = (int) (addresses.get(doc + 1L) - startOffset);
+                        bytesSlice.readBytes(startOffset, bytes.bytes, 0, bytes.length);
+                        return bytes;
+                    }
+                };
+            }
+        } else {
+            // sparse
+            final IndexedDISI disi = new IndexedDISI(
+                data,
+                entry.docsWithFieldOffset,
+                entry.docsWithFieldLength,
+                entry.jumpTableEntryCount,
+                entry.denseRankPower,
+                entry.numDocsWithField
+            );
+            if (entry.minLength == entry.maxLength) {
+                // fixed length
+                final int length = entry.maxLength;
+                return new SparseBinaryDocValues(disi) {
+                    final BytesRef bytes = new BytesRef(new byte[length], 0, length);
+
+                    @Override
+                    public BytesRef binaryValue() throws IOException {
+                        bytesSlice.readBytes((long) disi.index() * length, bytes.bytes, 0, length);
+                        return bytes;
+                    }
+                };
+            } else {
+                // variable length
+                final RandomAccessInput addressesData = this.data.randomAccessSlice(entry.addressesOffset, entry.addressesLength);
+                final LongValues addresses = DirectMonotonicReader.getInstance(entry.addressesMeta, addressesData, merging);
+                return new SparseBinaryDocValues(disi) {
+                    final BytesRef bytes = new BytesRef(new byte[entry.maxLength], 0, entry.maxLength);
+
+                    @Override
+                    public BytesRef binaryValue() throws IOException {
+                        final int index = disi.index();
+                        long startOffset = addresses.get(index);
+                        bytes.length = (int) (addresses.get(index + 1L) - startOffset);
+                        bytesSlice.readBytes(startOffset, bytes.bytes, 0, bytes.length);
+                        return bytes;
+                    }
+                };
+            }
+        }
+    }
+
+    private abstract static class DenseBinaryDocValues extends BinaryDocValues {
+
+        final int maxDoc;
+        int doc = -1;
+
+        DenseBinaryDocValues(int maxDoc) {
+            this.maxDoc = maxDoc;
+        }
+
+        @Override
+        public int nextDoc() throws IOException {
+            return advance(doc + 1);
+        }
+
+        @Override
+        public int docID() {
+            return doc;
+        }
+
+        @Override
+        public long cost() {
+            return maxDoc;
+        }
+
+        @Override
+        public int advance(int target) throws IOException {
+            if (target >= maxDoc) {
+                return doc = NO_MORE_DOCS;
+            }
+            return doc = target;
+        }
+
+        @Override
+        public boolean advanceExact(int target) throws IOException {
+            doc = target;
+            return true;
+        }
+    }
+
+    private abstract static class SparseBinaryDocValues extends BinaryDocValues {
+
+        final IndexedDISI disi;
+
+        SparseBinaryDocValues(IndexedDISI disi) {
+            this.disi = disi;
+        }
+
+        @Override
+        public int nextDoc() throws IOException {
+            return disi.nextDoc();
+        }
+
+        @Override
+        public int docID() {
+            return disi.docID();
+        }
+
+        @Override
+        public long cost() {
+            return disi.cost();
+        }
+
+        @Override
+        public int advance(int target) throws IOException {
+            return disi.advance(target);
+        }
+
+        @Override
+        public boolean advanceExact(int target) throws IOException {
+            return disi.advanceExact(target);
+        }
+    }
+
+    @Override
+    public SortedDocValues getSorted(FieldInfo field) throws IOException {
+        SortedEntry entry = sorted.get(field.number);
+        return getSorted(entry);
+    }
+
+    private SortedDocValues getSorted(SortedEntry entry) throws IOException {
+        final NumericDocValues ords = getNumeric(entry.ordsEntry, entry.termsDictEntry.termsDictSize);
+        return new BaseSortedDocValues(entry) {
+
+            @Override
+            public int ordValue() throws IOException {
+                return (int) ords.longValue();
+            }
+
+            @Override
+            public boolean advanceExact(int target) throws IOException {
+                return ords.advanceExact(target);
+            }
+
+            @Override
+            public int docID() {
+                return ords.docID();
+            }
+
+            @Override
+            public int nextDoc() throws IOException {
+                return ords.nextDoc();
+            }
+
+            @Override
+            public int advance(int target) throws IOException {
+                return ords.advance(target);
+            }
+
+            @Override
+            public long cost() {
+                return ords.cost();
+            }
+        };
+    }
+
+    abstract class BaseSortedDocValues extends SortedDocValues {
+
+        final SortedEntry entry;
+        final TermsEnum termsEnum;
+
+        BaseSortedDocValues(SortedEntry entry) throws IOException {
+            this.entry = entry;
+            this.termsEnum = termsEnum();
+        }
+
+        @Override
+        public int getValueCount() {
+            return Math.toIntExact(entry.termsDictEntry.termsDictSize);
+        }
+
+        @Override
+        public BytesRef lookupOrd(int ord) throws IOException {
+            termsEnum.seekExact(ord);
+            return termsEnum.term();
+        }
+
+        @Override
+        public int lookupTerm(BytesRef key) throws IOException {
+            TermsEnum.SeekStatus status = termsEnum.seekCeil(key);
+            return switch (status) {
+                case FOUND -> Math.toIntExact(termsEnum.ord());
+                default -> Math.toIntExact(-1L - termsEnum.ord());
+            };
+        }
+
+        @Override
+        public TermsEnum termsEnum() throws IOException {
+            return new TermsDict(entry.termsDictEntry, data, merging);
+        }
+    }
+
+    abstract static class BaseSortedSetDocValues extends SortedSetDocValues {
+
+        final SortedSetEntry entry;
+        final IndexInput data;
+        final boolean merging;
+        final TermsEnum termsEnum;
+
+        BaseSortedSetDocValues(SortedSetEntry entry, IndexInput data, boolean merging) throws IOException {
+            this.entry = entry;
+            this.data = data;
+            this.merging = merging;
+            this.termsEnum = termsEnum();
+        }
+
+        @Override
+        public long getValueCount() {
+            return entry.termsDictEntry.termsDictSize;
+        }
+
+        @Override
+        public BytesRef lookupOrd(long ord) throws IOException {
+            termsEnum.seekExact(ord);
+            return termsEnum.term();
+        }
+
+        @Override
+        public long lookupTerm(BytesRef key) throws IOException {
+            TermsEnum.SeekStatus status = termsEnum.seekCeil(key);
+            return switch (status) {
+                case FOUND -> termsEnum.ord();
+                default -> -1L - termsEnum.ord();
+            };
+        }
+
+        @Override
+        public TermsEnum termsEnum() throws IOException {
+            return new TermsDict(entry.termsDictEntry, data, merging);
+        }
+    }
+
+    private static class TermsDict extends BaseTermsEnum {
+        static final int LZ4_DECOMPRESSOR_PADDING = 7;
+
+        final TermsDictEntry entry;
+        final LongValues blockAddresses;
+        final IndexInput bytes;
+        final long blockMask;
+        final LongValues indexAddresses;
+        final RandomAccessInput indexBytes;
+        final BytesRef term;
+        long ord = -1;
+
+        BytesRef blockBuffer = null;
+        ByteArrayDataInput blockInput = null;
+        long currentCompressedBlockStart = -1;
+        long currentCompressedBlockEnd = -1;
+
+        TermsDict(TermsDictEntry entry, IndexInput data, boolean merging) throws IOException {
+            this.entry = entry;
+            RandomAccessInput addressesSlice = data.randomAccessSlice(entry.termsAddressesOffset, entry.termsAddressesLength);
+            blockAddresses = DirectMonotonicReader.getInstance(entry.termsAddressesMeta, addressesSlice, merging);
+            bytes = data.slice("terms", entry.termsDataOffset, entry.termsDataLength);
+            blockMask = (1L << TERMS_DICT_BLOCK_LZ4_SHIFT) - 1;
+            RandomAccessInput indexAddressesSlice = data.randomAccessSlice(
+                entry.termsIndexAddressesOffset,
+                entry.termsIndexAddressesLength
+            );
+            indexAddresses = DirectMonotonicReader.getInstance(entry.termsIndexAddressesMeta, indexAddressesSlice, merging);
+            indexBytes = data.randomAccessSlice(entry.termsIndexOffset, entry.termsIndexLength);
+            term = new BytesRef(entry.maxTermLength);
+
+            // add the max term length for the dictionary
+            // add 7 padding bytes can help decompression run faster.
+            int bufferSize = entry.maxBlockLength + entry.maxTermLength + LZ4_DECOMPRESSOR_PADDING;
+            blockBuffer = new BytesRef(new byte[bufferSize], 0, bufferSize);
+        }
+
+        @Override
+        public BytesRef next() throws IOException {
+            if (++ord >= entry.termsDictSize) {
+                return null;
+            }
+
+            if ((ord & blockMask) == 0L) {
+                decompressBlock();
+            } else {
+                DataInput input = blockInput;
+                final int token = Byte.toUnsignedInt(input.readByte());
+                int prefixLength = token & 0x0F;
+                int suffixLength = 1 + (token >>> 4);
+                if (prefixLength == 15) {
+                    prefixLength += input.readVInt();
+                }
+                if (suffixLength == 16) {
+                    suffixLength += input.readVInt();
+                }
+                term.length = prefixLength + suffixLength;
+                input.readBytes(term.bytes, prefixLength, suffixLength);
+            }
+            return term;
+        }
+
+        @Override
+        public void seekExact(long ord) throws IOException {
+            if (ord < 0 || ord >= entry.termsDictSize) {
+                throw new IndexOutOfBoundsException();
+            }
+            // Signed shift since ord is -1 when the terms enum is not positioned
+            final long currentBlockIndex = this.ord >> TERMS_DICT_BLOCK_LZ4_SHIFT;
+            final long blockIndex = ord >> TERMS_DICT_BLOCK_LZ4_SHIFT;
+            if (ord < this.ord || blockIndex != currentBlockIndex) {
+                // The looked up ord is before the current ord or belongs to a different block, seek again
+                final long blockAddress = blockAddresses.get(blockIndex);
+                bytes.seek(blockAddress);
+                this.ord = (blockIndex << TERMS_DICT_BLOCK_LZ4_SHIFT) - 1;
+            }
+            // Scan to the looked up ord
+            while (this.ord < ord) {
+                next();
+            }
+        }
+
+        private BytesRef getTermFromIndex(long index) throws IOException {
+            assert index >= 0 && index <= (entry.termsDictSize - 1) >>> entry.termsDictIndexShift;
+            final long start = indexAddresses.get(index);
+            term.length = (int) (indexAddresses.get(index + 1) - start);
+            indexBytes.readBytes(start, term.bytes, 0, term.length);
+            return term;
+        }
+
+        private long seekTermsIndex(BytesRef text) throws IOException {
+            long lo = 0L;
+            long hi = (entry.termsDictSize - 1) >> entry.termsDictIndexShift;
+            while (lo <= hi) {
+                final long mid = (lo + hi) >>> 1;
+                getTermFromIndex(mid);
+                final int cmp = term.compareTo(text);
+                if (cmp <= 0) {
+                    lo = mid + 1;
+                } else {
+                    hi = mid - 1;
+                }
+            }
+
+            assert hi < 0 || getTermFromIndex(hi).compareTo(text) <= 0;
+            assert hi == ((entry.termsDictSize - 1) >> entry.termsDictIndexShift) || getTermFromIndex(hi + 1).compareTo(text) > 0;
+
+            return hi;
+        }
+
+        private BytesRef getFirstTermFromBlock(long block) throws IOException {
+            assert block >= 0 && block <= (entry.termsDictSize - 1) >>> TERMS_DICT_BLOCK_LZ4_SHIFT;
+            final long blockAddress = blockAddresses.get(block);
+            bytes.seek(blockAddress);
+            term.length = bytes.readVInt();
+            bytes.readBytes(term.bytes, 0, term.length);
+            return term;
+        }
+
+        private long seekBlock(BytesRef text) throws IOException {
+            long index = seekTermsIndex(text);
+            if (index == -1L) {
+                return -1L;
+            }
+
+            long ordLo = index << entry.termsDictIndexShift;
+            long ordHi = Math.min(entry.termsDictSize, ordLo + (1L << entry.termsDictIndexShift)) - 1L;
+
+            long blockLo = ordLo >>> TERMS_DICT_BLOCK_LZ4_SHIFT;
+            long blockHi = ordHi >>> TERMS_DICT_BLOCK_LZ4_SHIFT;
+
+            while (blockLo <= blockHi) {
+                final long blockMid = (blockLo + blockHi) >>> 1;
+                getFirstTermFromBlock(blockMid);
+                final int cmp = term.compareTo(text);
+                if (cmp <= 0) {
+                    blockLo = blockMid + 1;
+                } else {
+                    blockHi = blockMid - 1;
+                }
+            }
+
+            assert blockHi < 0 || getFirstTermFromBlock(blockHi).compareTo(text) <= 0;
+            assert blockHi == ((entry.termsDictSize - 1) >>> TERMS_DICT_BLOCK_LZ4_SHIFT)
+                || getFirstTermFromBlock(blockHi + 1).compareTo(text) > 0;
+
+            return blockHi;
+        }
+
+        @Override
+        public SeekStatus seekCeil(BytesRef text) throws IOException {
+            final long block = seekBlock(text);
+            if (block == -1) {
+                // before the first term, or empty terms dict
+                if (entry.termsDictSize == 0) {
+                    ord = 0;
+                    return SeekStatus.END;
+                } else {
+                    seekExact(0L);
+                    return SeekStatus.NOT_FOUND;
+                }
+            }
+            final long blockAddress = blockAddresses.get(block);
+            this.ord = block << TERMS_DICT_BLOCK_LZ4_SHIFT;
+            bytes.seek(blockAddress);
+            decompressBlock();
+
+            while (true) {
+                int cmp = term.compareTo(text);
+                if (cmp == 0) {
+                    return SeekStatus.FOUND;
+                } else if (cmp > 0) {
+                    return SeekStatus.NOT_FOUND;
+                }
+                if (next() == null) {
+                    return SeekStatus.END;
+                }
+            }
+        }
+
+        private void decompressBlock() throws IOException {
+            // The first term is kept uncompressed, so no need to decompress block if only
+            // look up the first term when doing seek block.
+            term.length = bytes.readVInt();
+            bytes.readBytes(term.bytes, 0, term.length);
+            long offset = bytes.getFilePointer();
+            if (offset < entry.termsDataLength - 1) {
+                // Avoid decompress again if we are reading a same block.
+                if (currentCompressedBlockStart != offset) {
+                    blockBuffer.offset = term.length;
+                    blockBuffer.length = bytes.readVInt();
+                    // Decompress the remaining of current block, using the first term as a dictionary
+                    System.arraycopy(term.bytes, 0, blockBuffer.bytes, 0, blockBuffer.offset);
+                    LZ4.decompress(bytes, blockBuffer.length, blockBuffer.bytes, blockBuffer.offset);
+                    currentCompressedBlockStart = offset;
+                    currentCompressedBlockEnd = bytes.getFilePointer();
+                } else {
+                    // Skip decompression but need to re-seek to block end.
+                    bytes.seek(currentCompressedBlockEnd);
+                }
+
+                // Reset the buffer.
+                blockInput = new ByteArrayDataInput(blockBuffer.bytes, blockBuffer.offset, blockBuffer.length);
+            }
+        }
+
+        @Override
+        public BytesRef term() throws IOException {
+            return term;
+        }
+
+        @Override
+        public long ord() throws IOException {
+            return ord;
+        }
+
+        @Override
+        public long totalTermFreq() throws IOException {
+            return -1L;
+        }
+
+        @Override
+        public PostingsEnum postings(PostingsEnum reuse, int flags) throws IOException {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public ImpactsEnum impacts(int flags) throws IOException {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public int docFreq() throws IOException {
+            throw new UnsupportedOperationException();
+        }
+    }
+
+    @Override
+    public SortedNumericDocValues getSortedNumeric(FieldInfo field) throws IOException {
+        SortedNumericEntry entry = sortedNumerics.get(field.number);
+        return getSortedNumeric(entry, -1);
+    }
+
+    @Override
+    public SortedSetDocValues getSortedSet(FieldInfo field) throws IOException {
+        SortedSetEntry entry = sortedSets.get(field.number);
+        if (entry.singleValueEntry != null) {
+            return DocValues.singleton(getSorted(entry.singleValueEntry));
+        }
+
+        SortedNumericEntry ordsEntry = entry.ordsEntry;
+        final SortedNumericDocValues ords = getSortedNumeric(ordsEntry, entry.termsDictEntry.termsDictSize);
+        return new BaseSortedSetDocValues(entry, data, merging) {
+
+            int i = 0;
+            int count = 0;
+            boolean set = false;
+
+            @Override
+            public long nextOrd() throws IOException {
+                if (set == false) {
+                    set = true;
+                    i = 0;
+                    count = ords.docValueCount();
+                }
+                assert i < count;
+                i++;
+                return ords.nextValue();
+            }
+
+            @Override
+            public int docValueCount() {
+                return ords.docValueCount();
+            }
+
+            @Override
+            public boolean advanceExact(int target) throws IOException {
+                set = false;
+                return ords.advanceExact(target);
+            }
+
+            @Override
+            public int docID() {
+                return ords.docID();
+            }
+
+            @Override
+            public int nextDoc() throws IOException {
+                set = false;
+                return ords.nextDoc();
+            }
+
+            @Override
+            public int advance(int target) throws IOException {
+                set = false;
+                return ords.advance(target);
+            }
+
+            @Override
+            public long cost() {
+                return ords.cost();
+            }
+        };
+    }
+
+    @Override
+    public DocValuesSkipper getSkipper(FieldInfo field) throws IOException {
+        final DocValuesSkipperEntry entry = skippers.get(field.number);
+
+        final IndexInput input = data.slice("doc value skipper", entry.offset, entry.length);
+        // Prefetch the first page of data. Following pages are expected to get prefetched through
+        // read-ahead.
+        if (input.length() > 0) {
+            input.prefetch(0, 1);
+        }
+        // TODO: should we write to disk the actual max level for this segment?
+        return new DocValuesSkipper() {
+            final int[] minDocID = new int[SKIP_INDEX_MAX_LEVEL];
+            final int[] maxDocID = new int[SKIP_INDEX_MAX_LEVEL];
+
+            {
+                for (int i = 0; i < SKIP_INDEX_MAX_LEVEL; i++) {
+                    minDocID[i] = maxDocID[i] = -1;
+                }
+            }
+
+            final long[] minValue = new long[SKIP_INDEX_MAX_LEVEL];
+            final long[] maxValue = new long[SKIP_INDEX_MAX_LEVEL];
+            final int[] docCount = new int[SKIP_INDEX_MAX_LEVEL];
+            int levels = 1;
+
+            @Override
+            public void advance(int target) throws IOException {
+                if (target > entry.maxDocId) {
+                    // skipper is exhausted
+                    for (int i = 0; i < SKIP_INDEX_MAX_LEVEL; i++) {
+                        minDocID[i] = maxDocID[i] = DocIdSetIterator.NO_MORE_DOCS;
+                    }
+                } else {
+                    // find next interval
+                    assert target > maxDocID[0] : "target must be bigger that current interval";
+                    while (true) {
+                        levels = input.readByte();
+                        assert levels <= SKIP_INDEX_MAX_LEVEL && levels > 0 : "level out of range [" + levels + "]";
+                        boolean valid = true;
+                        // check if current interval is competitive or we can jump to the next position
+                        for (int level = levels - 1; level >= 0; level--) {
+                            if ((maxDocID[level] = input.readInt()) < target) {
+                                input.skipBytes(SKIP_INDEX_JUMP_LENGTH_PER_LEVEL[level]); // the jump for the level
+                                valid = false;
+                                break;
+                            }
+                            minDocID[level] = input.readInt();
+                            maxValue[level] = input.readLong();
+                            minValue[level] = input.readLong();
+                            docCount[level] = input.readInt();
+                        }
+                        if (valid) {
+                            // adjust levels
+                            while (levels < SKIP_INDEX_MAX_LEVEL && maxDocID[levels] >= target) {
+                                levels++;
+                            }
+                            break;
+                        }
+                    }
+                }
+            }
+
+            @Override
+            public int numLevels() {
+                return levels;
+            }
+
+            @Override
+            public int minDocID(int level) {
+                return minDocID[level];
+            }
+
+            @Override
+            public int maxDocID(int level) {
+                return maxDocID[level];
+            }
+
+            @Override
+            public long minValue(int level) {
+                return minValue[level];
+            }
+
+            @Override
+            public long maxValue(int level) {
+                return maxValue[level];
+            }
+
+            @Override
+            public int docCount(int level) {
+                return docCount[level];
+            }
+
+            @Override
+            public long minValue() {
+                return entry.minValue;
+            }
+
+            @Override
+            public long maxValue() {
+                return entry.maxValue;
+            }
+
+            @Override
+            public int docCount() {
+                return entry.docCount;
+            }
+        };
+    }
+
+    @Override
+    public void checkIntegrity() throws IOException {
+        CodecUtil.checksumEntireFile(data);
+    }
+
+    @Override
+    public void close() throws IOException {
+        data.close();
+    }
+
+    private void readFields(IndexInput meta, FieldInfos infos) throws IOException {
+        for (int fieldNumber = meta.readInt(); fieldNumber != -1; fieldNumber = meta.readInt()) {
+            FieldInfo info = infos.fieldInfo(fieldNumber);
+            if (info == null) {
+                throw new CorruptIndexException("Invalid field number: " + fieldNumber, meta);
+            }
+            byte type = meta.readByte();
+            if (info.docValuesSkipIndexType() != DocValuesSkipIndexType.NONE) {
+                skippers.put(info.number, readDocValueSkipperMeta(meta));
+            }
+            if (type == ES819TSDBDocValuesFormat.NUMERIC) {
+                numerics.put(info.number, readNumeric(meta));
+            } else if (type == ES819TSDBDocValuesFormat.BINARY) {
+                binaries.put(info.number, readBinary(meta));
+            } else if (type == ES819TSDBDocValuesFormat.SORTED) {
+                sorted.put(info.number, readSorted(meta));
+            } else if (type == ES819TSDBDocValuesFormat.SORTED_SET) {
+                sortedSets.put(info.number, readSortedSet(meta));
+            } else if (type == ES819TSDBDocValuesFormat.SORTED_NUMERIC) {
+                sortedNumerics.put(info.number, readSortedNumeric(meta));
+            } else {
+                throw new CorruptIndexException("invalid type: " + type, meta);
+            }
+        }
+    }
+
+    private static NumericEntry readNumeric(IndexInput meta) throws IOException {
+        NumericEntry entry = new NumericEntry();
+        readNumeric(meta, entry);
+        return entry;
+    }
+
+    private static DocValuesSkipperEntry readDocValueSkipperMeta(IndexInput meta) throws IOException {
+        long offset = meta.readLong();
+        long length = meta.readLong();
+        long maxValue = meta.readLong();
+        long minValue = meta.readLong();
+        int docCount = meta.readInt();
+        int maxDocID = meta.readInt();
+
+        return new DocValuesSkipperEntry(offset, length, minValue, maxValue, docCount, maxDocID);
+    }
+
+    private static void readNumeric(IndexInput meta, NumericEntry entry) throws IOException {
+        entry.numValues = meta.readLong();
+        // Change compared to ES87TSDBDocValuesProducer:
+        entry.numDocsWithField = meta.readInt();
+        if (entry.numValues > 0) {
+            final int indexBlockShift = meta.readInt();
+            // Special case, -1 means there are no blocks, so no need to load the metadata for it
+            // -1 is written when there the cardinality of a field is exactly one.
+            if (indexBlockShift != -1) {
+                entry.indexMeta = DirectMonotonicReader.loadMeta(
+                    meta,
+                    1 + ((entry.numValues - 1) >>> ES819TSDBDocValuesFormat.NUMERIC_BLOCK_SHIFT),
+                    indexBlockShift
+                );
+            }
+            entry.indexOffset = meta.readLong();
+            entry.indexLength = meta.readLong();
+            entry.valuesOffset = meta.readLong();
+            entry.valuesLength = meta.readLong();
+        }
+        // Change compared to ES87TSDBDocValuesProducer:
+        entry.docsWithFieldOffset = meta.readLong();
+        entry.docsWithFieldLength = meta.readLong();
+        entry.jumpTableEntryCount = meta.readShort();
+        entry.denseRankPower = meta.readByte();
+    }
+
+    private BinaryEntry readBinary(IndexInput meta) throws IOException {
+        final BinaryEntry entry = new BinaryEntry();
+        entry.dataOffset = meta.readLong();
+        entry.dataLength = meta.readLong();
+        entry.docsWithFieldOffset = meta.readLong();
+        entry.docsWithFieldLength = meta.readLong();
+        entry.jumpTableEntryCount = meta.readShort();
+        entry.denseRankPower = meta.readByte();
+        entry.numDocsWithField = meta.readInt();
+        entry.minLength = meta.readInt();
+        entry.maxLength = meta.readInt();
+        if (entry.minLength < entry.maxLength) {
+            entry.addressesOffset = meta.readLong();
+
+            // Old count of uncompressed addresses
+            long numAddresses = entry.numDocsWithField + 1L;
+
+            final int blockShift = meta.readVInt();
+            entry.addressesMeta = DirectMonotonicReader.loadMeta(meta, numAddresses, blockShift);
+            entry.addressesLength = meta.readLong();
+        }
+        return entry;
+    }
+
+    private static SortedNumericEntry readSortedNumeric(IndexInput meta) throws IOException {
+        SortedNumericEntry entry = new SortedNumericEntry();
+        readSortedNumeric(meta, entry);
+        return entry;
+    }
+
+    private static SortedNumericEntry readSortedNumeric(IndexInput meta, SortedNumericEntry entry) throws IOException {
+        readNumeric(meta, entry);
+        // We don't read numDocsWithField here any more.
+        if (entry.numDocsWithField != entry.numValues) {
+            entry.addressesOffset = meta.readLong();
+            final int blockShift = meta.readVInt();
+            entry.addressesMeta = DirectMonotonicReader.loadMeta(meta, entry.numDocsWithField + 1, blockShift);
+            entry.addressesLength = meta.readLong();
+        }
+        return entry;
+    }
+
+    private SortedEntry readSorted(IndexInput meta) throws IOException {
+        SortedEntry entry = new SortedEntry();
+        entry.ordsEntry = new NumericEntry();
+        readNumeric(meta, entry.ordsEntry);
+        entry.termsDictEntry = new TermsDictEntry();
+        readTermDict(meta, entry.termsDictEntry);
+        return entry;
+    }
+
+    private SortedSetEntry readSortedSet(IndexInput meta) throws IOException {
+        SortedSetEntry entry = new SortedSetEntry();
+        byte multiValued = meta.readByte();
+        switch (multiValued) {
+            case 0: // singlevalued
+                entry.singleValueEntry = readSorted(meta);
+                return entry;
+            case 1: // multivalued
+                break;
+            default:
+                throw new CorruptIndexException("Invalid multiValued flag: " + multiValued, meta);
+        }
+        entry.ordsEntry = new SortedNumericEntry();
+        readSortedNumeric(meta, entry.ordsEntry);
+        entry.termsDictEntry = new TermsDictEntry();
+        readTermDict(meta, entry.termsDictEntry);
+        return entry;
+    }
+
+    private static void readTermDict(IndexInput meta, TermsDictEntry entry) throws IOException {
+        entry.termsDictSize = meta.readVLong();
+        final int blockShift = meta.readInt();
+        final long addressesSize = (entry.termsDictSize + (1L << TERMS_DICT_BLOCK_LZ4_SHIFT) - 1) >>> TERMS_DICT_BLOCK_LZ4_SHIFT;
+        entry.termsAddressesMeta = DirectMonotonicReader.loadMeta(meta, addressesSize, blockShift);
+        entry.maxTermLength = meta.readInt();
+        entry.maxBlockLength = meta.readInt();
+        entry.termsDataOffset = meta.readLong();
+        entry.termsDataLength = meta.readLong();
+        entry.termsAddressesOffset = meta.readLong();
+        entry.termsAddressesLength = meta.readLong();
+        entry.termsDictIndexShift = meta.readInt();
+        final long indexSize = (entry.termsDictSize + (1L << entry.termsDictIndexShift) - 1) >>> entry.termsDictIndexShift;
+        entry.termsIndexAddressesMeta = DirectMonotonicReader.loadMeta(meta, 1 + indexSize, blockShift);
+        entry.termsIndexOffset = meta.readLong();
+        entry.termsIndexLength = meta.readLong();
+        entry.termsIndexAddressesOffset = meta.readLong();
+        entry.termsIndexAddressesLength = meta.readLong();
+    }
+
+    private abstract static class NumericValues {
+        abstract long advance(long index) throws IOException;
+    }
+
+    private NumericDocValues getNumeric(NumericEntry entry, long maxOrd) throws IOException {
+        if (entry.docsWithFieldOffset == -2) {
+            // empty
+            return DocValues.emptyNumeric();
+        }
+
+        if (maxOrd == 1) {
+            // Special case for maxOrd 1, no need to read blocks and use ordinal 0 as only value
+            if (entry.docsWithFieldOffset == -1) {
+                // Special case when all docs have a value
+                return new NumericDocValues() {
+
+                    private final int maxDoc = ES819TSDBDocValuesProducer.this.maxDoc;
+                    private int doc = -1;
+
+                    @Override
+                    public long longValue() {
+                        // Only one ordinal!
+                        return 0L;
+                    }
+
+                    @Override
+                    public int docID() {
+                        return doc;
+                    }
+
+                    @Override
+                    public int nextDoc() throws IOException {
+                        return advance(doc + 1);
+                    }
+
+                    @Override
+                    public int advance(int target) throws IOException {
+                        if (target >= maxDoc) {
+                            return doc = NO_MORE_DOCS;
+                        }
+                        return doc = target;
+                    }
+
+                    @Override
+                    public boolean advanceExact(int target) {
+                        doc = target;
+                        return true;
+                    }
+
+                    @Override
+                    public long cost() {
+                        return maxDoc;
+                    }
+                };
+            } else {
+                final IndexedDISI disi = new IndexedDISI(
+                    data,
+                    entry.docsWithFieldOffset,
+                    entry.docsWithFieldLength,
+                    entry.jumpTableEntryCount,
+                    entry.denseRankPower,
+                    entry.numValues
+                );
+                return new NumericDocValues() {
+
+                    @Override
+                    public int advance(int target) throws IOException {
+                        return disi.advance(target);
+                    }
+
+                    @Override
+                    public boolean advanceExact(int target) throws IOException {
+                        return disi.advanceExact(target);
+                    }
+
+                    @Override
+                    public int nextDoc() throws IOException {
+                        return disi.nextDoc();
+                    }
+
+                    @Override
+                    public int docID() {
+                        return disi.docID();
+                    }
+
+                    @Override
+                    public long cost() {
+                        return disi.cost();
+                    }
+
+                    @Override
+                    public long longValue() {
+                        return 0L;
+                    }
+                };
+            }
+        }
+
+        // NOTE: we could make this a bit simpler by reusing #getValues but this
+        // makes things slower.
+
+        final RandomAccessInput indexSlice = data.randomAccessSlice(entry.indexOffset, entry.indexLength);
+        final DirectMonotonicReader indexReader = DirectMonotonicReader.getInstance(entry.indexMeta, indexSlice, merging);
+        final IndexInput valuesData = data.slice("values", entry.valuesOffset, entry.valuesLength);
+
+        final int bitsPerOrd = maxOrd >= 0 ? PackedInts.bitsRequired(maxOrd - 1) : -1;
+        if (entry.docsWithFieldOffset == -1) {
+            // dense
+            return new NumericDocValues() {
+
+                private final int maxDoc = ES819TSDBDocValuesProducer.this.maxDoc;
+                private int doc = -1;
+                private final TSDBDocValuesEncoder decoder = new TSDBDocValuesEncoder(ES819TSDBDocValuesFormat.NUMERIC_BLOCK_SIZE);
+                private long currentBlockIndex = -1;
+                private final long[] currentBlock = new long[ES819TSDBDocValuesFormat.NUMERIC_BLOCK_SIZE];
+
+                @Override
+                public int docID() {
+                    return doc;
+                }
+
+                @Override
+                public int nextDoc() throws IOException {
+                    return advance(doc + 1);
+                }
+
+                @Override
+                public int advance(int target) throws IOException {
+                    if (target >= maxDoc) {
+                        return doc = NO_MORE_DOCS;
+                    }
+                    return doc = target;
+                }
+
+                @Override
+                public boolean advanceExact(int target) {
+                    doc = target;
+                    return true;
+                }
+
+                @Override
+                public long cost() {
+                    return maxDoc;
+                }
+
+                @Override
+                public long longValue() throws IOException {
+                    final int index = doc;
+                    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;
+                        if (maxOrd >= 0) {
+                            decoder.decodeOrdinals(valuesData, currentBlock, bitsPerOrd);
+                        } else {
+                            decoder.decode(valuesData, currentBlock);
+                        }
+                    }
+                    return currentBlock[blockInIndex];
+                }
+            };
+        } else {
+            final IndexedDISI disi = new IndexedDISI(
+                data,
+                entry.docsWithFieldOffset,
+                entry.docsWithFieldLength,
+                entry.jumpTableEntryCount,
+                entry.denseRankPower,
+                entry.numValues
+            );
+            return new NumericDocValues() {
+
+                private final TSDBDocValuesEncoder decoder = new TSDBDocValuesEncoder(ES819TSDBDocValuesFormat.NUMERIC_BLOCK_SIZE);
+                private long currentBlockIndex = -1;
+                private final long[] currentBlock = new long[ES819TSDBDocValuesFormat.NUMERIC_BLOCK_SIZE];
+
+                @Override
+                public int advance(int target) throws IOException {
+                    return disi.advance(target);
+                }
+
+                @Override
+                public boolean advanceExact(int target) throws IOException {
+                    return disi.advanceExact(target);
+                }
+
+                @Override
+                public int nextDoc() throws IOException {
+                    return disi.nextDoc();
+                }
+
+                @Override
+                public int docID() {
+                    return disi.docID();
+                }
+
+                @Override
+                public long cost() {
+                    return disi.cost();
+                }
+
+                @Override
+                public long longValue() throws IOException {
+                    final int index = disi.index();
+                    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;
+                        if (maxOrd >= 0) {
+                            decoder.decodeOrdinals(valuesData, currentBlock, bitsPerOrd);
+                        } else {
+                            decoder.decode(valuesData, currentBlock);
+                        }
+                    }
+                    return currentBlock[blockInIndex];
+                }
+            };
+        }
+    }
+
+    private NumericValues getValues(NumericEntry entry, final long maxOrd) throws IOException {
+        assert entry.numValues > 0;
+        final RandomAccessInput indexSlice = data.randomAccessSlice(entry.indexOffset, entry.indexLength);
+        final DirectMonotonicReader indexReader = DirectMonotonicReader.getInstance(entry.indexMeta, indexSlice, merging);
+
+        final IndexInput valuesData = data.slice("values", entry.valuesOffset, entry.valuesLength);
+        final int bitsPerOrd = maxOrd >= 0 ? PackedInts.bitsRequired(maxOrd - 1) : -1;
+        return new NumericValues() {
+
+            private final TSDBDocValuesEncoder decoder = new TSDBDocValuesEncoder(ES819TSDBDocValuesFormat.NUMERIC_BLOCK_SIZE);
+            private long currentBlockIndex = -1;
+            private final long[] currentBlock = new long[ES819TSDBDocValuesFormat.NUMERIC_BLOCK_SIZE];
+
+            @Override
+            long advance(long index) throws IOException {
+                final long blockIndex = index >>> ES819TSDBDocValuesFormat.NUMERIC_BLOCK_SHIFT;
+                final int blockInIndex = (int) (index & ES819TSDBDocValuesFormat.NUMERIC_BLOCK_MASK);
+                if (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;
+                    if (bitsPerOrd >= 0) {
+                        decoder.decodeOrdinals(valuesData, currentBlock, bitsPerOrd);
+                    } else {
+                        decoder.decode(valuesData, currentBlock);
+                    }
+                }
+                return currentBlock[blockInIndex];
+            }
+        };
+    }
+
+    private SortedNumericDocValues getSortedNumeric(SortedNumericEntry entry, long maxOrd) throws IOException {
+        if (entry.numValues == entry.numDocsWithField) {
+            return DocValues.singleton(getNumeric(entry, maxOrd));
+        }
+
+        final RandomAccessInput addressesInput = data.randomAccessSlice(entry.addressesOffset, entry.addressesLength);
+        final LongValues addresses = DirectMonotonicReader.getInstance(entry.addressesMeta, addressesInput, merging);
+
+        final NumericValues values = getValues(entry, maxOrd);
+
+        if (entry.docsWithFieldOffset == -1) {
+            // dense
+            return new SortedNumericDocValues() {
+
+                int doc = -1;
+                long start, end;
+                int count;
+
+                @Override
+                public int nextDoc() throws IOException {
+                    return advance(doc + 1);
+                }
+
+                @Override
+                public int docID() {
+                    return doc;
+                }
+
+                @Override
+                public long cost() {
+                    return maxDoc;
+                }
+
+                @Override
+                public int advance(int target) throws IOException {
+                    if (target >= maxDoc) {
+                        return doc = NO_MORE_DOCS;
+                    }
+                    start = addresses.get(target);
+                    end = addresses.get(target + 1L);
+                    count = (int) (end - start);
+                    return doc = target;
+                }
+
+                @Override
+                public boolean advanceExact(int target) throws IOException {
+                    start = addresses.get(target);
+                    end = addresses.get(target + 1L);
+                    count = (int) (end - start);
+                    doc = target;
+                    return true;
+                }
+
+                @Override
+                public long nextValue() throws IOException {
+                    return values.advance(start++);
+                }
+
+                @Override
+                public int docValueCount() {
+                    return count;
+                }
+            };
+        } else {
+            // sparse
+            final IndexedDISI disi = new IndexedDISI(
+                data,
+                entry.docsWithFieldOffset,
+                entry.docsWithFieldLength,
+                entry.jumpTableEntryCount,
+                entry.denseRankPower,
+                entry.numDocsWithField
+            );
+            return new SortedNumericDocValues() {
+
+                boolean set;
+                long start, end;
+                int count;
+
+                @Override
+                public int nextDoc() throws IOException {
+                    set = false;
+                    return disi.nextDoc();
+                }
+
+                @Override
+                public int docID() {
+                    return disi.docID();
+                }
+
+                @Override
+                public long cost() {
+                    return disi.cost();
+                }
+
+                @Override
+                public int advance(int target) throws IOException {
+                    set = false;
+                    return disi.advance(target);
+                }
+
+                @Override
+                public boolean advanceExact(int target) throws IOException {
+                    set = false;
+                    return disi.advanceExact(target);
+                }
+
+                @Override
+                public long nextValue() throws IOException {
+                    set();
+                    return values.advance(start++);
+                }
+
+                @Override
+                public int docValueCount() {
+                    set();
+                    return count;
+                }
+
+                private void set() {
+                    if (set == false) {
+                        final int index = disi.index();
+                        start = addresses.get(index);
+                        end = addresses.get(index + 1L);
+                        count = (int) (end - start);
+                        set = true;
+                    }
+                }
+            };
+        }
+    }
+
+    private record DocValuesSkipperEntry(long offset, long length, long minValue, long maxValue, int docCount, int maxDocId) {}
+
+    private static class NumericEntry {
+        long docsWithFieldOffset;
+        long docsWithFieldLength;
+        short jumpTableEntryCount;
+        byte denseRankPower;
+        long numValues;
+        // Change compared to ES87TSDBDocValuesProducer:
+        int numDocsWithField;
+        long indexOffset;
+        long indexLength;
+        DirectMonotonicReader.Meta indexMeta;
+        long valuesOffset;
+        long valuesLength;
+    }
+
+    private static class BinaryEntry {
+        long dataOffset;
+        long dataLength;
+        long docsWithFieldOffset;
+        long docsWithFieldLength;
+        short jumpTableEntryCount;
+        byte denseRankPower;
+        int numDocsWithField;
+        int minLength;
+        int maxLength;
+        long addressesOffset;
+        long addressesLength;
+        DirectMonotonicReader.Meta addressesMeta;
+    }
+
+    private static class SortedNumericEntry extends NumericEntry {
+        DirectMonotonicReader.Meta addressesMeta;
+        long addressesOffset;
+        long addressesLength;
+    }
+
+    private static class SortedEntry {
+        NumericEntry ordsEntry;
+        TermsDictEntry termsDictEntry;
+    }
+
+    private static class SortedSetEntry {
+        SortedEntry singleValueEntry;
+        SortedNumericEntry ordsEntry;
+        TermsDictEntry termsDictEntry;
+    }
+
+    private static class TermsDictEntry {
+        long termsDictSize;
+        DirectMonotonicReader.Meta termsAddressesMeta;
+        int maxTermLength;
+        long termsDataOffset;
+        long termsDataLength;
+        long termsAddressesOffset;
+        long termsAddressesLength;
+        int termsDictIndexShift;
+        DirectMonotonicReader.Meta termsIndexAddressesMeta;
+        long termsIndexOffset;
+        long termsIndexLength;
+        long termsIndexAddressesOffset;
+        long termsIndexAddressesLength;
+
+        int maxBlockLength;
+    }
+
+}

server/src/main/resources/META-INF/services/org.apache.lucene.codecs.DocValuesFormat

@@ -1 +1,2 @@
 org.elasticsearch.index.codec.tsdb.ES87TSDBDocValuesFormat
+org.elasticsearch.index.codec.tsdb.es819.ES819TSDBDocValuesFormat

server/src/test/java/org/elasticsearch/index/codec/tsdb/DocValuesCodecDuelTests.java

@@ -24,6 +24,8 @@ import org.apache.lucene.tests.index.ForceMergePolicy;
 import org.apache.lucene.tests.index.RandomIndexWriter;
 import org.apache.lucene.tests.util.TestUtil;
 import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.index.codec.tsdb.ES87TSDBDocValuesFormatTests.TestES87TSDBDocValuesFormat;
+import org.elasticsearch.index.codec.tsdb.es819.ES819TSDBDocValuesFormat;
 import org.elasticsearch.test.ESTestCase;
 
 import java.io.IOException;
@@ -39,6 +41,7 @@ public class DocValuesCodecDuelTests extends ESTestCase {
     private static final String FIELD_4 = "number_field_4";
     private static final String FIELD_5 = "binary_field_5";
 
+    @SuppressWarnings("checkstyle:LineLength")
     public void testDuel() throws IOException {
         try (var baselineDirectory = newDirectory(); var contenderDirectory = newDirectory()) {
             int numDocs = randomIntBetween(256, 32768);
@@ -48,7 +51,9 @@ public class DocValuesCodecDuelTests extends ESTestCase {
             baselineConfig.setMergePolicy(mergePolicy);
             baselineConfig.setCodec(TestUtil.alwaysDocValuesFormat(new Lucene90DocValuesFormat()));
             var contenderConf = newIndexWriterConfig();
-            contenderConf.setCodec(TestUtil.alwaysDocValuesFormat(new ES87TSDBDocValuesFormat()));
+            contenderConf.setCodec(
+                TestUtil.alwaysDocValuesFormat(rarely() ? new TestES87TSDBDocValuesFormat() : new ES819TSDBDocValuesFormat())
+            );
             contenderConf.setMergePolicy(mergePolicy);
 
             try (

server/src/test/java/org/elasticsearch/index/codec/tsdb/ES87TSDBDocValuesFormatTests.java

@@ -11,6 +11,7 @@ package org.elasticsearch.index.codec.tsdb;
 
 import org.apache.lucene.analysis.Analyzer;
 import org.apache.lucene.codecs.Codec;
+import org.apache.lucene.codecs.DocValuesConsumer;
 import org.apache.lucene.document.Document;
 import org.apache.lucene.document.Field;
 import org.apache.lucene.document.SortedDocValuesField;
@@ -22,6 +23,7 @@ import org.apache.lucene.index.IndexReader;
 import org.apache.lucene.index.IndexWriter;
 import org.apache.lucene.index.IndexWriterConfig;
 import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.index.SegmentWriteState;
 import org.apache.lucene.index.SortedDocValues;
 import org.apache.lucene.index.SortedNumericDocValues;
 import org.apache.lucene.index.SortedSetDocValues;
@@ -48,7 +50,23 @@ public class ES87TSDBDocValuesFormatTests extends BaseDocValuesFormatTestCase {
 
     private static final int NUM_DOCS = 10;
 
-    private final Codec codec = TestUtil.alwaysDocValuesFormat(new ES87TSDBDocValuesFormat());
+    static class TestES87TSDBDocValuesFormat extends ES87TSDBDocValuesFormat {
+
+        TestES87TSDBDocValuesFormat() {
+            super();
+        }
+
+        TestES87TSDBDocValuesFormat(int skipIndexIntervalSize) {
+            super(skipIndexIntervalSize);
+        }
+
+        @Override
+        public DocValuesConsumer fieldsConsumer(SegmentWriteState state) throws IOException {
+            return new ES87TSDBDocValuesConsumer(state, skipIndexIntervalSize, DATA_CODEC, DATA_EXTENSION, META_CODEC, META_EXTENSION);
+        }
+    }
+
+    private final Codec codec = TestUtil.alwaysDocValuesFormat(new TestES87TSDBDocValuesFormat());
 
     @Override
     protected Codec getCodec() {

server/src/test/java/org/elasticsearch/index/codec/tsdb/ES87TSDBDocValuesFormatVariableSkipIntervalTests.java

@@ -32,7 +32,7 @@ public class ES87TSDBDocValuesFormatVariableSkipIntervalTests extends BaseDocVal
     @Override
     protected Codec getCodec() {
         // small interval size to test with many intervals
-        return TestUtil.alwaysDocValuesFormat(new ES87TSDBDocValuesFormat(random().nextInt(4, 16)));
+        return TestUtil.alwaysDocValuesFormat(new ES87TSDBDocValuesFormatTests.TestES87TSDBDocValuesFormat(random().nextInt(4, 16)));
     }
 
     public void testSkipIndexIntervalSize() {

server/src/test/java/org/elasticsearch/index/codec/tsdb/TsdbDocValueBwcTests.java

@@ -0,0 +1,274 @@
+/*
+ * 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;
+
+import org.apache.lucene.codecs.Codec;
+import org.apache.lucene.codecs.DocValuesProducer;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.NumericDocValuesField;
+import org.apache.lucene.document.SortedDocValuesField;
+import org.apache.lucene.document.SortedNumericDocValuesField;
+import org.apache.lucene.document.SortedSetDocValuesField;
+import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.DocValues;
+import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.IndexWriterConfig;
+import org.apache.lucene.index.LogByteSizeMergePolicy;
+import org.apache.lucene.index.MultiDocValues;
+import org.apache.lucene.index.NoMergePolicy;
+import org.apache.lucene.index.SegmentReader;
+import org.apache.lucene.search.Sort;
+import org.apache.lucene.search.SortField;
+import org.apache.lucene.search.SortedNumericSortField;
+import org.apache.lucene.tests.util.TestUtil;
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.cluster.metadata.DataStream;
+import org.elasticsearch.core.SuppressForbidden;
+import org.elasticsearch.index.codec.tsdb.ES87TSDBDocValuesFormatTests.TestES87TSDBDocValuesFormat;
+import org.elasticsearch.index.codec.tsdb.es819.ES819TSDBDocValuesFormat;
+import org.elasticsearch.test.ESTestCase;
+import org.hamcrest.Matchers;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.Arrays;
+import java.util.Locale;
+import java.util.Map;
+
+public class TsdbDocValueBwcTests extends ESTestCase {
+
+    public void testMixedIndex() throws Exception {
+        Codec oldCodec = TestUtil.alwaysDocValuesFormat(new TestES87TSDBDocValuesFormat());
+        Codec newCodec = TestUtil.alwaysDocValuesFormat(new ES819TSDBDocValuesFormat());
+        testMixedIndex(oldCodec, newCodec);
+    }
+
+    void testMixedIndex(Codec oldCodec, Codec newCodec) throws IOException, NoSuchFieldException, IllegalAccessException,
+        ClassNotFoundException {
+        String timestampField = "@timestamp";
+        String hostnameField = "host.name";
+        long baseTimestamp = 1704067200000L;
+        int numRounds = 4 + random().nextInt(8);
+        int numDocsPerRound = 64 + random().nextInt(128);
+        int numDocs = numRounds * numDocsPerRound;
+
+        try (var dir = newDirectory()) {
+            long counter1 = 0;
+            long[] gauge1Values = new long[] { 2, 4, 6, 8, 10, 12, 14, 16 };
+            String[] tags = new String[] { "tag_1", "tag_2", "tag_3", "tag_4", "tag_5", "tag_6", "tag_7", "tag_8" };
+            try (var iw = new IndexWriter(dir, getTimeSeriesIndexWriterConfig(hostnameField, timestampField, oldCodec))) {
+                long timestamp = baseTimestamp;
+                for (int i = 0; i < numRounds; i++) {
+                    int r = random().nextInt(10);
+                    for (int j = 0; j < numDocsPerRound; j++) {
+                        var d = new Document();
+                        // host in reverse, otherwise merging will detect that segments are already ordered and will use sequential docid
+                        // merger:
+                        String hostName = String.format(Locale.ROOT, "host-%03d", numRounds - i);
+                        d.add(new SortedDocValuesField(hostnameField, new BytesRef(hostName)));
+                        // Index sorting doesn't work with NumericDocValuesField:
+                        d.add(new SortedNumericDocValuesField(timestampField, timestamp++));
+
+                        if (r % 10 < 8) {
+                            // Most of the time store counter:
+                            d.add(new NumericDocValuesField("counter_1", counter1++));
+                        }
+
+                        if (r % 10 == 5) {
+                            // sometimes no values
+                        } else if (r % 10 > 5) {
+                            // often single value:
+                            d.add(new SortedNumericDocValuesField("gauge_1", gauge1Values[j % gauge1Values.length]));
+                            d.add(new SortedSetDocValuesField("tags", new BytesRef(tags[j % tags.length])));
+                        } else {
+                            // otherwise multiple values:
+                            int numValues = 2 + random().nextInt(4);
+                            for (int k = 0; k < numValues; k++) {
+                                d.add(new SortedNumericDocValuesField("gauge_1", gauge1Values[(j + k) % gauge1Values.length]));
+                                d.add(new SortedSetDocValuesField("tags", new BytesRef(tags[(j + k) % tags.length])));
+                            }
+                        }
+                        iw.addDocument(d);
+                    }
+                    iw.commit();
+                }
+            }
+            // Check documents before force merge:
+            try (var iw = new IndexWriter(dir, getTimeSeriesIndexWriterConfig(hostnameField, timestampField, newCodec))) {
+                try (var reader = DirectoryReader.open(iw)) {
+                    assertOldDocValuesFormatVersion(reader);
+
+                    var hostNameDV = MultiDocValues.getSortedValues(reader, hostnameField);
+                    assertNotNull(hostNameDV);
+                    var timestampDV = MultiDocValues.getSortedNumericValues(reader, timestampField);
+                    assertNotNull(timestampDV);
+                    var counterOneDV = MultiDocValues.getNumericValues(reader, "counter_1");
+                    if (counterOneDV == null) {
+                        counterOneDV = DocValues.emptyNumeric();
+                    }
+                    var gaugeOneDV = MultiDocValues.getSortedNumericValues(reader, "gauge_1");
+                    if (gaugeOneDV == null) {
+                        gaugeOneDV = DocValues.emptySortedNumeric();
+                    }
+                    var tagsDV = MultiDocValues.getSortedSetValues(reader, "tags");
+                    if (tagsDV == null) {
+                        tagsDV = DocValues.emptySortedSet();
+                    }
+                    for (int i = 0; i < numDocs; i++) {
+                        assertEquals(i, hostNameDV.nextDoc());
+                        String actualHostName = hostNameDV.lookupOrd(hostNameDV.ordValue()).utf8ToString();
+                        assertTrue("unexpected host name:" + actualHostName, actualHostName.startsWith("host-"));
+
+                        assertEquals(i, timestampDV.nextDoc());
+                        long timestamp = timestampDV.nextValue();
+                        long lowerBound = baseTimestamp;
+                        long upperBound = baseTimestamp + numDocs;
+                        assertTrue(
+                            "unexpected timestamp [" + timestamp + "], expected between [" + lowerBound + "] and [" + upperBound + "]",
+                            timestamp >= lowerBound && timestamp < upperBound
+                        );
+                        if (counterOneDV.advanceExact(i)) {
+                            long counterOneValue = counterOneDV.longValue();
+                            assertTrue("unexpected counter [" + counterOneValue + "]", counterOneValue >= 0 && counterOneValue < counter1);
+                        }
+                        if (gaugeOneDV.advanceExact(i)) {
+                            for (int j = 0; j < gaugeOneDV.docValueCount(); j++) {
+                                long value = gaugeOneDV.nextValue();
+                                assertTrue("unexpected gauge [" + value + "]", Arrays.binarySearch(gauge1Values, value) >= 0);
+                            }
+                        }
+                        if (tagsDV.advanceExact(i)) {
+                            for (int j = 0; j < tagsDV.docValueCount(); j++) {
+                                long ordinal = tagsDV.nextOrd();
+                                String actualTag = tagsDV.lookupOrd(ordinal).utf8ToString();
+                                assertTrue("unexpected tag [" + actualTag + "]", Arrays.binarySearch(tags, actualTag) >= 0);
+                            }
+                        }
+                    }
+                }
+            }
+
+            var iwc = getTimeSeriesIndexWriterConfig(hostnameField, timestampField, newCodec);
+            iwc.setMergePolicy(new LogByteSizeMergePolicy());
+            try (var iw = new IndexWriter(dir, iwc)) {
+                iw.forceMerge(1);
+                // Check documents after force merge:
+                try (var reader = DirectoryReader.open(iw)) {
+                    assertEquals(1, reader.leaves().size());
+                    assertEquals(numDocs, reader.maxDoc());
+                    assertNewDocValuesFormatVersion(reader);
+                    var leaf = reader.leaves().get(0).reader();
+                    var hostNameDV = leaf.getSortedDocValues(hostnameField);
+                    assertNotNull(hostNameDV);
+                    var timestampDV = DocValues.unwrapSingleton(leaf.getSortedNumericDocValues(timestampField));
+                    assertNotNull(timestampDV);
+                    var counterOneDV = leaf.getNumericDocValues("counter_1");
+                    if (counterOneDV == null) {
+                        counterOneDV = DocValues.emptyNumeric();
+                    }
+                    var gaugeOneDV = leaf.getSortedNumericDocValues("gauge_1");
+                    if (gaugeOneDV == null) {
+                        gaugeOneDV = DocValues.emptySortedNumeric();
+                    }
+                    var tagsDV = leaf.getSortedSetDocValues("tags");
+                    if (tagsDV == null) {
+                        tagsDV = DocValues.emptySortedSet();
+                    }
+                    for (int i = 0; i < numDocs; i++) {
+                        assertEquals(i, hostNameDV.nextDoc());
+                        String actualHostName = hostNameDV.lookupOrd(hostNameDV.ordValue()).utf8ToString();
+                        assertTrue("unexpected host name:" + actualHostName, actualHostName.startsWith("host-"));
+
+                        assertEquals(i, timestampDV.nextDoc());
+                        long timestamp = timestampDV.longValue();
+                        long lowerBound = baseTimestamp;
+                        long upperBound = baseTimestamp + numDocs;
+                        assertTrue(
+                            "unexpected timestamp [" + timestamp + "], expected between [" + lowerBound + "] and [" + upperBound + "]",
+                            timestamp >= lowerBound && timestamp < upperBound
+                        );
+                        if (counterOneDV.advanceExact(i)) {
+                            long counterOneValue = counterOneDV.longValue();
+                            assertTrue("unexpected counter [" + counterOneValue + "]", counterOneValue >= 0 && counterOneValue < counter1);
+                        }
+                        if (gaugeOneDV.advanceExact(i)) {
+                            for (int j = 0; j < gaugeOneDV.docValueCount(); j++) {
+                                long value = gaugeOneDV.nextValue();
+                                assertTrue("unexpected gauge [" + value + "]", Arrays.binarySearch(gauge1Values, value) >= 0);
+                            }
+                        }
+                        if (tagsDV.advanceExact(i)) {
+                            for (int j = 0; j < tagsDV.docValueCount(); j++) {
+                                long ordinal = tagsDV.nextOrd();
+                                String actualTag = tagsDV.lookupOrd(ordinal).utf8ToString();
+                                assertTrue("unexpected tag [" + actualTag + "]", Arrays.binarySearch(tags, actualTag) >= 0);
+                            }
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    private IndexWriterConfig getTimeSeriesIndexWriterConfig(String hostnameField, String timestampField, Codec codec) {
+        var config = new IndexWriterConfig();
+        config.setIndexSort(
+            new Sort(
+                new SortField(hostnameField, SortField.Type.STRING, false),
+                new SortedNumericSortField(timestampField, SortField.Type.LONG, true)
+            )
+        );
+        config.setLeafSorter(DataStream.TIMESERIES_LEAF_READERS_SORTER);
+        // avoids the usage of ES87TSDBDocValuesProducer while indexing using old codec:
+        // (The per field format encodes the dv codec name and that then loads the current dv codec)
+        config.setMergePolicy(NoMergePolicy.INSTANCE);
+        config.setCodec(codec);
+        return config;
+    }
+
+    // A hacky way to figure out whether doc values format is written in what version. Need to use reflection, because
+    // PerFieldDocValuesFormat hides the doc values formats it wraps.
+    private static void assertOldDocValuesFormatVersion(DirectoryReader reader) throws NoSuchFieldException, IllegalAccessException,
+        IOException {
+        for (var leafReaderContext : reader.leaves()) {
+            var leaf = (SegmentReader) leafReaderContext.reader();
+            var dvReader = leaf.getDocValuesReader();
+            var field = getFormatsFieldFromPerFieldFieldsReader(dvReader.getClass());
+            Map<?, ?> formats = (Map<?, ?>) field.get(dvReader);
+            var tsdbDvReader = (DocValuesProducer) formats.get("ES87TSDB_0");
+            tsdbDvReader.checkIntegrity();
+            assertThat(tsdbDvReader, Matchers.instanceOf(ES87TSDBDocValuesProducer.class));
+        }
+    }
+
+    private static void assertNewDocValuesFormatVersion(DirectoryReader reader) throws NoSuchFieldException, IllegalAccessException,
+        IOException, ClassNotFoundException {
+        for (var leafReaderContext : reader.leaves()) {
+            var leaf = (SegmentReader) leafReaderContext.reader();
+            var dvReader = leaf.getDocValuesReader();
+            var field = getFormatsFieldFromPerFieldFieldsReader(dvReader.getClass());
+            Map<?, ?> formats = (Map<?, ?>) field.get(dvReader);
+            var tsdbDvReader = (DocValuesProducer) formats.get("ES819TSDB_0");
+            tsdbDvReader.checkIntegrity();
+            assertThat(
+                tsdbDvReader,
+                Matchers.instanceOf(Class.forName("org.elasticsearch.index.codec.tsdb.es819.ES819TSDBDocValuesProducer"))
+            );
+        }
+    }
+
+    @SuppressForbidden(reason = "access violation required in order to read private field for this test")
+    private static Field getFormatsFieldFromPerFieldFieldsReader(Class<?> c) throws NoSuchFieldException {
+        var field = c.getDeclaredField("formats");
+        field.setAccessible(true);
+        return field;
+    }
+
+}

server/src/test/java/org/elasticsearch/index/codec/tsdb/es819/ES819TSDBDocValuesFormatTests.java

@@ -0,0 +1,25 @@
+/*
+ * 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.codecs.Codec;
+import org.apache.lucene.tests.util.TestUtil;
+import org.elasticsearch.index.codec.tsdb.ES87TSDBDocValuesFormatTests;
+
+public class ES819TSDBDocValuesFormatTests extends ES87TSDBDocValuesFormatTests {
+
+    private final Codec codec = TestUtil.alwaysDocValuesFormat(new ES819TSDBDocValuesFormat());
+
+    @Override
+    protected Codec getCodec() {
+        return codec;
+    }
+
+}

server/src/test/java/org/elasticsearch/index/codec/tsdb/es819/ES819TSDBDocValuesFormatVariableSkipIntervalTests.java

@@ -0,0 +1,31 @@
+/*
+ * 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.codecs.Codec;
+import org.apache.lucene.tests.util.TestUtil;
+import org.elasticsearch.index.codec.tsdb.ES87TSDBDocValuesFormatVariableSkipIntervalTests;
+
+/** Tests ES819TSDBDocValuesFormat with custom skipper interval size. */
+public class ES819TSDBDocValuesFormatVariableSkipIntervalTests extends ES87TSDBDocValuesFormatVariableSkipIntervalTests {
+
+    @Override
+    protected Codec getCodec() {
+        // small interval size to test with many intervals
+        return TestUtil.alwaysDocValuesFormat(new ES819TSDBDocValuesFormat(random().nextInt(4, 16)));
+    }
+
+    public void testSkipIndexIntervalSize() {
+        IllegalArgumentException ex = expectThrows(
+            IllegalArgumentException.class,
+            () -> new ES819TSDBDocValuesFormat(random().nextInt(Integer.MIN_VALUE, 2))
+        );
+        assertTrue(ex.getMessage().contains("skipIndexIntervalSize must be > 1"));
+    }
+}