Merge branch 'main' into pem-csv-rate-tests

Author: pabloem

.buildkite/pipelines/pull-request-performance-benchmark.yml

@@ -0,0 +1,3 @@
+steps:
+  - label: ":pipeline: Generate steps"
+    command: bash .buildkite/scripts/generate-pr-performance-benchmark.sh | buildkite-agent pipeline upload

.buildkite/pipelines/pull-request/performance-benchmark.yml

@@ -27,7 +27,8 @@
       "set_commit_status": false,
       "build_on_commit": false,
       "build_on_comment": true,
-      "trigger_comment_regex": "^(buildkite|@elastic(search)?machine) benchmark this with (?<benchmark>\\w+)( please)?$"
+      "target_branch": "main",
+      "trigger_comment_regex": "^(buildkite|@elastic(search)?machine) benchmark this with (?<benchmark>\\S+)( please)?$"
     }
   ]
 }

.buildkite/pull-requests.json

@@ -0,0 +1,40 @@
+#!/bin/bash
+
+set -euo pipefail
+
+env_id_baseline=$(python3 -c 'import uuid; print(uuid.uuid4())')
+env_id_contender=$(python3 -c 'import uuid; print(uuid.uuid4())')
+merge_base=$(git merge-base "${GITHUB_PR_TARGET_BRANCH}" HEAD)
+
+buildkite-agent meta-data set pr_comment:custom-body:body \
+  "This build attempted two ${GITHUB_PR_COMMENT_VAR_BENCHMARK} benchmarks to evaluate performance impact of this PR."
+buildkite-agent meta-data set pr_comment:custom-baseline:head \
+  "* Baseline: ${merge_base} (env ID ${env_id_baseline})"
+buildkite-agent meta-data set pr_comment:custom-contender:head \
+  "* Contender: ${GITHUB_PR_TRIGGERED_SHA} (env ID ${env_id_contender})"
+
+cat << _EOF_
+steps:
+  - label: Trigger baseline benchmark
+    trigger: elasticsearch-performance-esbench-pr
+    build:
+      message: Baseline benchmark for PR${GITHUB_PR_NUMBER}
+      branch: master
+      env:
+        CONFIGURATION_NAME: ${GITHUB_PR_COMMENT_VAR_BENCHMARK}
+        ENV_ID: ${env_id_baseline}
+        REVISION: ${merge_base}
+  - label: Trigger contender benchmark
+    trigger: elasticsearch-performance-esbench-pr
+    build:
+      message: Contender benchmark for PR${GITHUB_PR_NUMBER}
+      branch: master
+      env:
+        CONFIGURATION_NAME: ${GITHUB_PR_COMMENT_VAR_BENCHMARK}
+        ENV_ID: ${env_id_contender}
+        ES_REPO_URL: https://github.com/${GITHUB_PR_OWNER}/${GITHUB_PR_REPO}.git
+        REVISION: ${GITHUB_PR_TRIGGERED_SHA}
+  - wait: ~
+  - label: Modify PR comment
+    command: buildkite-agent meta-data set pr_comment:custom-comparison:head "* [Benchmark results](<https://esbench-metrics.kb.us-east-2.aws.elastic-cloud.com:9243/app/dashboards#/view/d9079962-5866-49ef-b9f5-145f2141cd31?_a=(query:(language:kuery,query:'user-tags.env-id:${env_id_baseline} or user-tags.env-id:${env_id_contender}'))>)"
+_EOF_

.buildkite/scripts/generate-pr-performance-benchmark.sh

@@ -1,4 +1,7 @@
 
+# claude
+.claude
+
 # intellij files
 .idea/
 *.iml

.gitignore

@@ -47,6 +47,7 @@ dependencies {
   api(project(':x-pack:plugin:core'))
   api(project(':x-pack:plugin:esql'))
   api(project(':x-pack:plugin:esql:compute'))
+  api(project(':x-pack:plugin:mapper-exponential-histogram'))
   implementation project(path: ':libs:native')
   implementation project(path: ':libs:simdvec')
   implementation project(path: ':libs:exponential-histogram')

benchmarks/build.gradle

@@ -9,11 +9,15 @@
 
 package org.elasticsearch.benchmark.exponentialhistogram;
 
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.exponentialhistogram.BucketIterator;
 import org.elasticsearch.exponentialhistogram.ExponentialHistogram;
 import org.elasticsearch.exponentialhistogram.ExponentialHistogramCircuitBreaker;
 import org.elasticsearch.exponentialhistogram.ExponentialHistogramGenerator;
 import org.elasticsearch.exponentialhistogram.ExponentialHistogramMerger;
+import org.elasticsearch.xpack.exponentialhistogram.CompressedExponentialHistogram;
+import org.elasticsearch.xpack.exponentialhistogram.IndexWithCount;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
 import org.openjdk.jmh.annotations.Fork;
@@ -27,6 +31,8 @@
 import org.openjdk.jmh.annotations.Threads;
 import org.openjdk.jmh.annotations.Warmup;
 
+import java.io.IOException;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Random;
 import java.util.concurrent.ThreadLocalRandom;
@@ -47,6 +53,9 @@ public class ExponentialHistogramMergeBench {
     @Param({ "0.01", "0.1", "0.25", "0.5", "1.0", "2.0" })
     double mergedHistoSizeFactor;
 
+    @Param({ "array-backed", "compressed" })
+    String histoImplementation;
+
     Random random;
     ExponentialHistogramMerger histoMerger;
 
@@ -81,16 +90,54 @@ public void setUp() {
                 bucketIndex += 1 + random.nextInt(bucketCount) % (Math.max(1, bucketCount / dataPointSize));
                 generator.add(Math.pow(1.001, bucketIndex));
             }
-            toMerge[i] = generator.getAndClear();
-            cnt = getBucketCount(toMerge[i]);
+            ExponentialHistogram histogram = generator.getAndClear();
+            cnt = getBucketCount(histogram);
             if (cnt < dataPointSize) {
-                throw new IllegalArgumentException("Expected bucket count to be " + dataPointSize + ", but was " + cnt);
+                throw new IllegalStateException("Expected bucket count to be " + dataPointSize + ", but was " + cnt);
+            }
+
+            if ("array-backed".equals(histoImplementation)) {
+                toMerge[i] = histogram;
+            } else if ("compressed".equals(histoImplementation)) {
+                toMerge[i] = asCompressedHistogram(histogram);
+            } else {
+                throw new IllegalArgumentException("Unknown implementation: " + histoImplementation);
             }
         }
 
         index = 0;
     }
 
+    private ExponentialHistogram asCompressedHistogram(ExponentialHistogram histogram) {
+        List<IndexWithCount> negativeBuckets = new ArrayList<>();
+        List<IndexWithCount> positiveBuckets = new ArrayList<>();
+
+        BucketIterator it = histogram.negativeBuckets().iterator();
+        while (it.hasNext()) {
+            negativeBuckets.add(new IndexWithCount(it.peekIndex(), it.peekCount()));
+            it.advance();
+        }
+        it = histogram.positiveBuckets().iterator();
+        while (it.hasNext()) {
+            positiveBuckets.add(new IndexWithCount(it.peekIndex(), it.peekCount()));
+            it.advance();
+        }
+
+        long totalCount = histogram.zeroBucket().count() + histogram.negativeBuckets().valueCount() + histogram.positiveBuckets()
+            .valueCount();
+        BytesStreamOutput histoBytes = new BytesStreamOutput();
+        try {
+            CompressedExponentialHistogram.writeHistogramBytes(histoBytes, histogram.scale(), negativeBuckets, positiveBuckets);
+            CompressedExponentialHistogram result = new CompressedExponentialHistogram();
+            BytesRef data = histoBytes.bytes().toBytesRef();
+            result.reset(histogram.zeroBucket().zeroThreshold(), totalCount, data);
+            return result;
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+
+    }
+
     private static int getBucketCount(ExponentialHistogram histo) {
         int cnt = 0;
         for (BucketIterator it : List.of(histo.negativeBuckets().iterator(), histo.positiveBuckets().iterator())) {

benchmarks/src/main/java/org/elasticsearch/benchmark/exponentialhistogram/ExponentialHistogramMergeBench.java

@@ -258,7 +258,7 @@ private static IndexWriterConfig createIndexWriterConfig(boolean optimizedMergeE
         );
         config.setLeafSorter(DataStream.TIMESERIES_LEAF_READERS_SORTER);
         config.setMergePolicy(new LogByteSizeMergePolicy());
-        var docValuesFormat = new ES819TSDBDocValuesFormat(4096, optimizedMergeEnabled);
+        var docValuesFormat = new ES819TSDBDocValuesFormat(4096, 512, optimizedMergeEnabled);
         config.setCodec(new Elasticsearch900Lucene101Codec() {
 
             @Override

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

@@ -1,7 +1,7 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionSha256Sum=443c9c8ee2ac1ee0e11881a40f2376d79c66386264a44b24a9f8ca67e633375f
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.14.2-all.zip
+distributionSha256Sum=f759b8dd5204e2e3fa4ca3e73f452f087153cf81bac9561eeb854229cc2c5365
+distributionUrl=https\://services.gradle.org/distributions/gradle-9.0.0-all.zip
 networkTimeout=10000
 validateDistributionUrl=true
 zipStoreBase=GRADLE_USER_HOME

build-tools-internal/gradle/wrapper/gradle-wrapper.properties

@@ -53,8 +53,10 @@ abstract class AbstractRestResourcesFuncTest extends AbstractGradleFuncTest {
         }
         """
 
-        subProject(":distribution:archives:integ-test-zip") << "configurations.create('extracted')\n"
-        subProject(":distribution:archives:integ-test-zip") << "configurations.create('default')\n"
+        subProject(":distribution:archives:integ-test-zip") << """
+apply plugin: 'base'
+configurations.create('extracted')
+"""
     }
 
     void setupRestResources(List<String> apis, List<String> tests = [], List<String> xpackTests = []) {