feat(tsdb): pipeline-based numeric codec with self-describing format and per-field encoding

benchmarks/build.gradle

@@ -104,11 +104,24 @@ tasks.named('test').configure {
   }
 }
 
+// NOTE: Copy NumericDataGenerators from server test sources so benchmarks can reuse
+// the same data patterns without adding a testFixtures dependency.
+def numericDataGenSource = "${project(':server').projectDir}/src/test/java/org/elasticsearch/index/codec/tsdb/pipeline/NumericDataGenerators.java"
+def numericDataGenDir = layout.buildDirectory.dir("generated/sources/numericDataGenerators")
+def copyNumericDataGenerators = tasks.register('copyNumericDataGenerators', Copy) {
+  inputs.file(numericDataGenSource)
+  from numericDataGenSource
+  into numericDataGenDir.map { it.dir("org/elasticsearch/index/codec/tsdb/pipeline") }
+  outputs.dir(numericDataGenDir)
+}
+sourceSets.main.java.srcDir numericDataGenDir
+tasks.named('compileJava').configure { dependsOn(copyNumericDataGenerators) }
+
 spotless {
   java {
     // IDEs can sometimes run annotation processors that leave files in
     // here, causing Spotless to complain. Even though this path ought not
     // to exist, exclude it anyway in order to avoid spurious failures.
-    targetExclude 'src/main/generated/**/*.java'
+    targetExclude 'src/main/generated/**/*.java', 'build/generated/**/*.java'
   }
 }

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

@@ -278,7 +278,8 @@ private static BlockLoader numericBlockLoader(WhereAndBaseName w, NumberFieldMap
             false,
             null,
             null,
-            false
+            false,
+            null
         ).blockLoader(blContext());
     }
 

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

@@ -0,0 +1,67 @@
+/*
+ * 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.elasticsearch.benchmark.index.codec.tsdb.internal.AbstractTSDBCodecBenchmark;
+import org.elasticsearch.benchmark.index.codec.tsdb.internal.AllStagesSupplier;
+import org.elasticsearch.benchmark.index.codec.tsdb.internal.DecodeBenchmark;
+import org.elasticsearch.benchmark.index.codec.tsdb.internal.ThroughputMetrics;
+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.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Benchmark for decoding data that triggers all encoding stages (Delta -> Offset -> GCD -> FOR).
+ */
+@Fork(value = 1)
+@Warmup(iterations = 3)
+@Measurement(iterations = 5)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@State(Scope.Benchmark)
+public class DecodeAllStagesBenchmark {
+    private static final int SEED = 17;
+
+    @Param({ "10" })
+    private int blocksPerInvocation;
+
+    private final AbstractTSDBCodecBenchmark decode;
+
+    public DecodeAllStagesBenchmark() {
+        this.decode = new DecodeBenchmark();
+    }
+
+    @Setup(Level.Trial)
+    public void setupTrial() throws IOException {
+        AllStagesSupplier supplier = AllStagesSupplier.builder(SEED, decode.getBlockSize()).build();
+        decode.setupTrial(supplier);
+        decode.setBlocksPerInvocation(blocksPerInvocation);
+        decode.run();
+    }
+
+    @Benchmark
+    public void throughput(Blackhole bh, ThroughputMetrics metrics) throws IOException {
+        decode.benchmark(bh);
+        metrics.recordOperation(decode.getBlockSize() * blocksPerInvocation, decode.getEncodedSize() * blocksPerInvocation);
+    }
+}

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

@@ -0,0 +1,109 @@
+/*
+ * 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.elasticsearch.benchmark.index.codec.tsdb.internal.DecodeBenchmark;
+import org.elasticsearch.benchmark.index.codec.tsdb.internal.NumericDataSupplierRegistry;
+import org.elasticsearch.benchmark.index.codec.tsdb.internal.ThroughputMetrics;
+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.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+// NOTE: Decoding benchmark using NumericDataGenerators double data sources.
+@Fork(value = 1)
+@Warmup(iterations = 3)
+@Measurement(iterations = 5)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@State(Scope.Benchmark)
+public class DecodeDoubleGeneratorsBenchmark {
+
+    @Param(
+        {
+            "constant-double",
+            "percentage-double",
+            "monotonic-double",
+            "gauge-double",
+            "realistic-gauge-double",
+            "sparse-gauge-double",
+            "random-double",
+            "stable-sensor-double",
+            "tiny-increment-double",
+            "steady-counter-double",
+            "burst-spike-double",
+            "zero-crossing-oscillation-double",
+            "step-with-spikes-double",
+            "counter-with-resets-double",
+            "quantized-double",
+            "sensor-2dp-double",
+            "temperature-1dp-double",
+            "financial-2dp-double",
+            "percentage-rounded-1dp-double",
+            "mixed-sign-double",
+            "step-hold-double",
+            "timestamp-as-double",
+            "counter-as-double",
+            "gauge-as-double",
+            "gcd-as-double",
+            "constant-as-double",
+            "random-as-double",
+            "decreasing-timestamp-as-double",
+            "small-as-double",
+            "timestamp-with-jitter-as-double" }
+    )
+    private String datasetName;
+
+    @Param({ "42" })
+    private String seed;
+
+    @Param({ "10" })
+    private int blocksPerInvocation;
+
+    private final DecodeBenchmark decode = new DecodeBenchmark();
+
+    @Setup(Level.Trial)
+    public void setupTrial() throws IOException {
+        final Map<String, Supplier<long[]>> registry = NumericDataSupplierRegistry.toMap(
+            NumericDataSupplierRegistry.doubleSuppliers(decode.getBlockSize(), Long.parseLong(seed))
+        );
+        if (registry.size() != 30) {
+            throw new IllegalStateException(
+                "Expected 30 double datasets but found " + registry.size() + ". Update @Param list. Available: " + registry.keySet()
+            );
+        }
+        final Supplier<long[]> supplier = registry.get(datasetName);
+        if (supplier == null) {
+            throw new IllegalArgumentException("Unknown double dataset: [" + datasetName + "]. Available: " + registry.keySet());
+        }
+        decode.setupTrial(supplier);
+        decode.setBlocksPerInvocation(blocksPerInvocation);
+    }
+
+    @Benchmark
+    public void throughput(Blackhole bh, ThroughputMetrics metrics) throws IOException {
+        decode.benchmark(bh);
+        metrics.recordOperation(decode.getBlockSize() * blocksPerInvocation, decode.getEncodedSize() * blocksPerInvocation);
+    }
+}

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

@@ -0,0 +1,84 @@
+/*
+ * 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.elasticsearch.benchmark.index.codec.tsdb.internal.DecodeBenchmark;
+import org.elasticsearch.benchmark.index.codec.tsdb.internal.NumericDataSupplierRegistry;
+import org.elasticsearch.benchmark.index.codec.tsdb.internal.ThroughputMetrics;
+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.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+// NOTE: Decoding benchmark using NumericDataGenerators long data sources.
+@Fork(value = 1)
+@Warmup(iterations = 3)
+@Measurement(iterations = 5)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@State(Scope.Benchmark)
+public class DecodeLongGeneratorsBenchmark {
+
+    @Param(
+        {
+            "timestamp",
+            "counter",
+            "gauge",
+            "gcd",
+            "constant",
+            "random",
+            "decreasing-timestamp",
+            "boundary",
+            "small",
+            "timestamp-with-jitter" }
+    )
+    private String datasetName;
+
+    @Param({ "42" })
+    private String seed;
+
+    @Param({ "10" })
+    private int blocksPerInvocation;
+
+    private final DecodeBenchmark decode = new DecodeBenchmark();
+
+    @Setup(Level.Trial)
+    public void setupTrial() throws IOException {
+        final Map<String, Supplier<long[]>> registry = NumericDataSupplierRegistry.toMap(
+            NumericDataSupplierRegistry.longSuppliers(decode.getBlockSize(), Long.parseLong(seed))
+        );
+        final Supplier<long[]> supplier = registry.get(datasetName);
+        if (supplier == null) {
+            throw new IllegalArgumentException("Unknown long dataset: [" + datasetName + "]. Available: " + registry.keySet());
+        }
+        decode.setupTrial(supplier);
+        decode.setBlocksPerInvocation(blocksPerInvocation);
+    }
+
+    @Benchmark
+    public void throughput(Blackhole bh, ThroughputMetrics metrics) throws IOException {
+        decode.benchmark(bh);
+        metrics.recordOperation(decode.getBlockSize() * blocksPerInvocation, decode.getEncodedSize() * blocksPerInvocation);
+    }
+}

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

@@ -0,0 +1,77 @@
+/*
+ * 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.elasticsearch.benchmark.index.codec.tsdb.internal.AbstractTSDBCodecBenchmark;
+import org.elasticsearch.benchmark.index.codec.tsdb.internal.AllStagesSupplier;
+import org.elasticsearch.benchmark.index.codec.tsdb.internal.CompressionMetrics;
+import org.elasticsearch.benchmark.index.codec.tsdb.internal.EncodeBenchmark;
+import org.elasticsearch.benchmark.index.codec.tsdb.internal.ThroughputMetrics;
+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.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Benchmark for encoding data that triggers all encoding stages (Delta -> Offset -> GCD -> FOR).
+ */
+@Fork(value = 1)
+@Warmup(iterations = 3)
+@Measurement(iterations = 5)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@State(Scope.Benchmark)
+public class EncodeAllStagesBenchmark {
+    private static final int SEED = 17;
+
+    @Param({ "10" })
+    private int blocksPerInvocation;
+
+    private final AbstractTSDBCodecBenchmark encode;
+
+    public EncodeAllStagesBenchmark() {
+        this.encode = new EncodeBenchmark();
+    }
+
+    @Setup(Level.Trial)
+    public void setupTrial() throws IOException {
+        AllStagesSupplier supplier = AllStagesSupplier.builder(SEED, encode.getBlockSize()).build();
+        encode.setupTrial(supplier);
+        encode.setBlocksPerInvocation(blocksPerInvocation);
+        encode.run();
+    }
+
+    @Benchmark
+    public void throughput(Blackhole bh, ThroughputMetrics metrics) throws IOException {
+        encode.benchmark(bh);
+        metrics.recordOperation(encode.getBlockSize() * blocksPerInvocation, encode.getEncodedSize() * blocksPerInvocation);
+    }
+
+    @Benchmark
+    @Warmup(iterations = 0)
+    @Measurement(iterations = 1)
+    @BenchmarkMode(Mode.SingleShotTime)
+    public void compression(Blackhole bh, CompressionMetrics metrics) throws IOException {
+        encode.benchmark(bh);
+        metrics.recordOperation(encode.getBlockSize(), encode.getEncodedSize(), 0);
+    }
+}