Add binary doc value compression with variable doc count blocks

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

@@ -127,12 +127,22 @@ public KnnVectorsFormat getKnnVectorsFormatForField(String field) {
     }
 
     public DocValuesFormat getDocValuesFormatForField(String field) {
-        if (useTSDBDocValuesFormat(field)) {
+        if (useTSDBDocValuesFormat(field) || isBinaryDocValueField(field)) {
             return tsdbDocValuesFormat;
         }
         return docValuesFormat;
     }
 
+    boolean isBinaryDocValueField(final String field) {
+        if (mapperService != null) {
+            Mapper mapper = mapperService.mappingLookup().getMapper(field);
+            if (mapper != null && "wildcard".equals(mapper.typeName())) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     boolean useTSDBDocValuesFormat(final String field) {
         if (excludeFields(field)) {
             return false;

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

@@ -0,0 +1,30 @@
+/*
+ * 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;
+
+public enum BinaryDVCompressionMode {
+
+    NO_COMPRESS((byte) 0),
+    COMPRESSED_WITH_ZSTD((byte) 1);
+
+    public final byte code;
+
+    BinaryDVCompressionMode(byte code) {
+        this.code = code;
+    }
+
+    public static BinaryDVCompressionMode fromMode(byte mode) {
+        return switch (mode) {
+            case 0 -> NO_COMPRESS;
+            case 1 -> COMPRESSED_WITH_ZSTD;
+            default -> throw new IllegalStateException("unknown compression mode [" + mode + "]");
+        };
+    }
+}

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

@@ -0,0 +1,111 @@
+/*
+ * 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.store.ChecksumIndexInput;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.store.IOContext;
+import org.apache.lucene.store.IndexOutput;
+import org.apache.lucene.util.IOUtils;
+import org.apache.lucene.util.packed.DirectMonotonicWriter;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/**
+ *  Like OffsetsAccumulator builds offsets and stores in a DirectMonotonicWriter. But write to temp file
+ *  rather than directly to a DirectMonotonicWriter because the number of values is unknown. If number of
+ *  values if known prefer OffsetsWriter.
+ */
+final class DelayedOffsetAccumulator implements Closeable {
+    private final Directory dir;
+    private final long startOffset;
+
+    private int numValues = 0;
+    private final IndexOutput tempOutput;
+    private final String suffix;
+
+    DelayedOffsetAccumulator(
+        Directory dir,
+        IOContext context,
+        IndexOutput data,
+        String suffix,
+        long startOffset
+    ) throws IOException {
+        this.dir = dir;
+        this.startOffset = startOffset;
+        this.suffix = suffix;
+
+        boolean success = false;
+        try {
+            tempOutput = dir.createTempOutput(data.getName(), suffix, context);
+            CodecUtil.writeHeader(
+                tempOutput,
+                ES819TSDBDocValuesFormat.META_CODEC + suffix,
+                ES819TSDBDocValuesFormat.VERSION_CURRENT
+            );
+            success = true;
+        }
+        finally {
+            if (success == false) {
+                IOUtils.closeWhileHandlingException(this); // self-close because constructor caller can't
+            }
+        }
+    }
+
+    public void addDoc(long delta) throws IOException {
+        tempOutput.writeVLong(delta);
+        numValues++;
+    }
+
+    public void build(IndexOutput meta, IndexOutput data) throws IOException {
+        CodecUtil.writeFooter(tempOutput);
+        IOUtils.close(tempOutput);
+
+        // write the offsets info to the meta file by reading from temp file
+        try (ChecksumIndexInput tempInput = dir.openChecksumInput(tempOutput.getName());) {
+            CodecUtil.checkHeader(
+                tempInput,
+                ES819TSDBDocValuesFormat.META_CODEC + suffix,
+                ES819TSDBDocValuesFormat.VERSION_CURRENT,
+                ES819TSDBDocValuesFormat.VERSION_CURRENT
+            );
+            Throwable priorE = null;
+            try {
+                final DirectMonotonicWriter writer = DirectMonotonicWriter.getInstance(
+                    meta,
+                    data,
+                    numValues + 1,
+                    ES819TSDBDocValuesFormat.DIRECT_MONOTONIC_BLOCK_SHIFT
+                );
+
+                long offset = startOffset;
+                writer.add(offset);
+                for (int i = 0; i < numValues; ++i) {
+                    offset += tempInput.readVLong();
+                    writer.add(offset);
+                }
+                writer.finish();
+            } catch (Throwable e) {
+                priorE = e;
+            } finally {
+                CodecUtil.checkFooter(tempInput, priorE);
+            }
+        }
+    }
+
+    @Override
+    public void close() throws IOException {
+        if (tempOutput != null) {
+            IOUtils.close(tempOutput, () -> dir.deleteFile(tempOutput.getName()));
+        }
+    }
+}