First step optimizing tsdb doc values codec merging.

The doc values codec iterates a few times over the doc value instance that needs to be written to disk. In case when merging and index sorting is enabled, this is much more expensive, as each time the doc values instance is iterated an expensive doc id sorting is performed (in order to get the doc ids in order of index sorting).

There are several reasons why the doc value instance is iterated multiple times:
* To compute stats (num values, number of docs with value) required for writing values to disk.
* To write bitset that indicate which documents have a value. (indexed disi, jump table)
* To write the actual values to disk.
* To write the addresses to disk (in case docs have multiple values)

This applies for numeric doc values, but also for the ordinals of sorted (set) doc values.

This PR addresses solving the first reason why doc value instance needs to be iterated. This is done only when in case of merging and when the segments to be merged with are also of type es87 doc values, codec version is the same and there are no deletes.

Author: martijnvg

benchmarks/src/main/java/org/elasticsearch/benchmark/index/codec/tsdb/TSDBDocValuesMergeBenchmark.java

@@ -0,0 +1,176 @@
+/*
+ * 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.benchmark.index.codec.tsdb;
+
+import org.apache.lucene.analysis.standard.StandardAnalyzer;
+import org.apache.lucene.codecs.DocValuesFormat;
+import org.apache.lucene.codecs.lucene101.Lucene101Codec;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.SortedDocValuesField;
+import org.apache.lucene.document.SortedNumericDocValuesField;
+import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.IndexWriterConfig;
+import org.apache.lucene.search.Sort;
+import org.apache.lucene.search.SortField;
+import org.apache.lucene.search.SortedNumericSortField;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.store.FSDirectory;
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.index.codec.tsdb.ES87TSDBDocValuesFormat;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.profile.AsyncProfiler;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Random;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+@BenchmarkMode(Mode.SampleTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@State(Scope.Benchmark)
+@Fork(1)
+@Threads(1)
+@Warmup(iterations = 0)
+@Measurement(iterations = 1)
+public class TSDBDocValuesMergeBenchmark {
+
+    @Param("13431204")
+    private int nDocs;
+
+    @Param("1000")
+    private int deltaTime;
+
+    @Param("42")
+    private int seed;
+
+    private static final String TIMESTAMP_FIELD = "@timestamp";
+    private static final String HOSTNAME_FIELD = "host.name";
+    private static final long BASE_TIMESTAMP = 1704067200000L;
+
+    private IndexWriter indexWriterWithoutOptimizedMerge;
+    private IndexWriter indexWriterWithOptimizedMerge;
+    private ExecutorService executorService;
+
+    public static void main(String[] args) throws RunnerException {
+        final Options options = new OptionsBuilder().include(TSDBDocValuesMergeBenchmark.class.getSimpleName())
+            .addProfiler(AsyncProfiler.class)
+            .build();
+
+        new Runner(options).run();
+    }
+
+    @Setup(Level.Trial)
+    public void setup() throws IOException {
+        executorService = Executors.newSingleThreadExecutor();
+
+        final Directory tempDirectoryWithoutDocValuesSkipper = FSDirectory.open(Files.createTempDirectory("temp1-"));
+        final Directory tempDirectoryWithDocValuesSkipper = FSDirectory.open(Files.createTempDirectory("temp2-"));
+
+        indexWriterWithoutOptimizedMerge = createIndex(tempDirectoryWithoutDocValuesSkipper, false);
+        indexWriterWithOptimizedMerge = createIndex(tempDirectoryWithDocValuesSkipper, true);
+    }
+
+    private IndexWriter createIndex(final Directory directory, final boolean optimizedMergeEnabled)
+        throws IOException {
+
+        final IndexWriterConfig config = new IndexWriterConfig(new StandardAnalyzer());
+        // NOTE: index sort config matching LogsDB's sort order
+        config.setIndexSort(
+            new Sort(
+                new SortField(HOSTNAME_FIELD, SortField.Type.STRING, false),
+                new SortedNumericSortField(TIMESTAMP_FIELD, SortField.Type.LONG, true)
+            )
+        );
+        ES87TSDBDocValuesFormat docValuesFormat = new ES87TSDBDocValuesFormat(4096, optimizedMergeEnabled);
+        config.setCodec(new Lucene101Codec() {
+
+            @Override
+            public DocValuesFormat getDocValuesFormatForField(String field) {
+                return docValuesFormat;
+            }
+        });
+
+        long counter1 = 0;
+        long counter2 = 10_000_000;
+        long[] gauge1Values = new long[] {2, 4, 6, 8, 10, 12, 14, 16};
+        long[] gauge2Values = new long[] {-2, -4, -6, -8, -10, -12, -14, -16};
+        int numHosts = 1000;
+
+        final Random random = new Random(seed);
+        IndexWriter indexWriter = new IndexWriter(directory, config);
+        for (int i = 0; i < nDocs; i++) {
+            final Document doc = new Document();
+
+            final int batchIndex = i / numHosts;
+            final String hostName = "host-" + batchIndex;
+            // Slightly vary the timestamp in each document
+            final long timestamp = BASE_TIMESTAMP + ((i % numHosts) * deltaTime) + random.nextInt(0, deltaTime);
+
+            doc.add(new SortedDocValuesField(HOSTNAME_FIELD, new BytesRef(hostName)));
+            doc.add(new SortedNumericDocValuesField(TIMESTAMP_FIELD, timestamp));
+            doc.add(new SortedNumericDocValuesField("counter_1", counter1++));
+            doc.add(new SortedNumericDocValuesField("counter_2", counter2++));
+            doc.add(new SortedNumericDocValuesField("gauge_1", gauge1Values[i % gauge1Values.length]));
+            doc.add(new SortedNumericDocValuesField("gauge_2", gauge2Values[i % gauge1Values.length]));
+
+            indexWriter.addDocument(doc);
+        }
+        indexWriter.commit();
+        return indexWriter;
+    }
+
+    @Benchmark
+    public void forceMergeWithoutOptimizedMerge() throws IOException {
+        forceMerge(indexWriterWithoutOptimizedMerge);
+    }
+
+    @Benchmark
+    public void forceMergeWithOptimizedMerge() throws IOException {
+        forceMerge(indexWriterWithOptimizedMerge);
+    }
+
+    private void forceMerge(final IndexWriter indexWriter) throws IOException {
+        indexWriter.forceMerge(1);
+    }
+
+    @TearDown(Level.Trial)
+    public void tearDown() {
+        if (executorService != null) {
+            executorService.shutdown();
+            try {
+                if (executorService.awaitTermination(30, TimeUnit.SECONDS) == false) {
+                    executorService.shutdownNow();
+                }
+            } catch (InterruptedException e) {
+                executorService.shutdownNow();
+                Thread.currentThread().interrupt();
+            }
+        }
+    }
+}