Merge branch 'main' into stateless-migrate-shards-availability-health-indicator

Author: masseyke

.buildkite/pull-requests.json

@@ -8,7 +8,7 @@
         "admin",
         "write"
       ],
-      "allowed_list": ["elastic-renovate-prod[bot]"],
+      "allowed_list": ["elastic-renovate-prod[bot]", "github-actions[bot]"],
       "set_commit_status": false,
       "build_on_commit": true,
       "build_on_comment": true,

.buildkite/scripts/dra-update-staging.sh

@@ -6,7 +6,7 @@ source .buildkite/scripts/branches.sh
 
 for BRANCH in "${BRANCHES[@]}"; do
   # Don't publish main branch to staging
-  if [[ "$BRANCH" == "main" ]]; then
+  if [[ "$BRANCH" == "main" || "$BRANCH" == "7.17" ]]; then
     continue
   fi
 

.github/updatecli/values.d/ironbank.yml

@@ -1,3 +1,12 @@
 config:
   - path: distribution/docker/src/docker/iron_bank
     dockerfile: ../dockerfiles/ironbank/Dockerfile
+
+pull_request:
+  labels:
+    - ":Delivery/Tooling"
+    - ":Delivery/Packaging"
+    - "auto-merge-without-approval"
+    - "dependencies"
+    - ">non-issue"
+

.github/updatecli/values.d/scm.yml

@@ -2,7 +2,7 @@ scm:
   enabled: true
   owner: elastic
   repository: elasticsearch
-  branch: main
+  branch: "main" # this value is overwritten by the GH workflow to match current branch
   commitusingapi: true
   # begin updatecli-compose policy values
   user: elasticmachine

.github/updatecli/values.d/updatecli-compose.yml

@@ -1,3 +1,9 @@
 spec:
   files:
-    - "updatecli-compose.yaml"
+    - "updatecli-compose.yaml"
+
+labels:
+  - ":Delivery/Tooling"
+  - "auto-merge-without-approval"
+  - "dependencies"
+  - ">non-issue"

.github/workflows/updatecli-compose.yml

@@ -10,15 +10,38 @@ permissions:
   contents: read
 
 jobs:
+  setup-matrix:
+    if: github.repository == 'elastic/elasticsearch'
+    runs-on: ubuntu-latest
+    outputs:
+      branches: ${{ steps.set-matrix.outputs.matrix }}
+    steps:
+      - uses: actions/checkout@v4
+      - id: set-matrix
+        run: |
+          BRANCHES=$(jq -c '[.branches[].branch]' branches.json)
+          echo "matrix=$BRANCHES" >> $GITHUB_OUTPUT
+
   compose:
+    needs: setup-matrix
     if: github.repository == 'elastic/elasticsearch'
     runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        branch: ${{ fromJson(needs.setup-matrix.outputs.branches) }}
     permissions:
       contents: write
       packages: read
       pull-requests: write
     steps:
       - uses: actions/checkout@v4
+        with:
+          ref: ${{ matrix.branch }}
+
+      - name: Update branch in SCM values
+        run: |
+          yq eval '.scm.branch = "${{ matrix.branch }}"' -i .github/updatecli/values.d/scm.yml
 
       - uses: docker/login-action@0d4c9c5ea7693da7b068278f7b52bda2a190a446 # v3.2.0
         with:
@@ -28,12 +51,7 @@ jobs:
 
       - uses: elastic/oblt-actions/updatecli/run@v1
         with:
-          command: --experimental compose diff
-        env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-
-      - uses: elastic/oblt-actions/updatecli/run@v1
-        with:
-          command: --experimental compose apply
+          # Runs in "--debug" mode to provide logs if the PR creation fails
+          command: --experimental compose apply --debug
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

.gitignore

@@ -73,6 +73,8 @@ testfixtures_shared/
 # Generated
 checkstyle_ide.xml
 x-pack/plugin/esql/src/main/generated-src/generated/
+x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/**/golden_tests/**/*.actual
+
 server/src/main/resources/transport/definitions/manifest.txt
 
 # JEnv

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

@@ -307,7 +307,7 @@ private static BlockLoader numericBlockLoader(WhereAndBaseName w, NumberFieldMap
     @OperationsPerInvocation(INDEX_SIZE)
     public void benchmark() {
         ValuesSourceReaderOperator op = new ValuesSourceReaderOperator(
-            new DriverContext(BigArrays.NON_RECYCLING_INSTANCE, blockFactory),
+            new DriverContext(BigArrays.NON_RECYCLING_INSTANCE, blockFactory, null),
             ByteSizeValue.ofMb(1).getBytes(),
             fields(name),
             new IndexedByShardIdFromSingleton<>(new ValuesSourceReaderOperator.ShardContext(reader, (sourcePaths) -> {

benchmarks/src/main/java/org/elasticsearch/benchmark/bytes/RecyclerBytesStreamOutputWriteBenchmark.java

@@ -0,0 +1,157 @@
+/*
+ * 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.bytes;
+
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.common.io.stream.RecyclerBytesStreamOutput;
+import org.elasticsearch.common.recycler.Recycler;
+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.OperationsPerInvocation;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+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 java.io.IOException;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+
+@BenchmarkMode(Mode.Throughput)
+@Warmup(time = 1)
+@Measurement(time = 1)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@State(Scope.Thread)
+@Fork(1)
+public class RecyclerBytesStreamOutputWriteBenchmark {
+
+    // large enough to negate stream reset
+    // and not too large to stay under stream limit of 2GB on worst case (5 bytes vint)
+    private static final int WRITES_PER_ITERATION = 400_000;
+    private static final int RANDOM_NUMS_PER_ITERATION = 1000;
+    private static final int PAGE_SIZE = 16384;
+
+    private ThreadLocalRandom random = ThreadLocalRandom.current();
+    private int[] vint1Byte;
+    private int[] vint2Bytes;
+    private int[] vint3Bytes;
+    private int[] vint4Bytes;
+    private int[] vint5Bytes;
+    private int[] vintNegBytes;
+    private RecyclerBytesStreamOutput output = new RecyclerBytesStreamOutput(new SinglePageStream());
+
+    private int randomVInt(int vIntByteSize, boolean isNeg) {
+        return (isNeg ? -1 : 1) * switch (vIntByteSize) {
+            case 1 -> random.nextInt(0, 1 << 7);
+            case 2 -> random.nextInt(1 << 7, 1 << 14);
+            case 3 -> random.nextInt(1 << 14, 1 << 21);
+            case 4 -> random.nextInt(1 << 21, 1 << 28);
+            case 5 -> random.nextInt(1 << 28, 1 << 30);
+            default -> throw new IllegalArgumentException("number of bytes must be between 1 and 5");
+        };
+    }
+
+    private int[] randomArray(int vIntByteSize, boolean isNeg) {
+        final var out = new int[RANDOM_NUMS_PER_ITERATION];
+        for (var i = 0; i < RANDOM_NUMS_PER_ITERATION; i++) {
+            out[i] = randomVInt(vIntByteSize, isNeg);
+        }
+        return out;
+    }
+
+    @Setup(Level.Iteration)
+    public void resetNums() {
+        vint1Byte = randomArray(1, false);
+        vint2Bytes = randomArray(2, false);
+        vint3Bytes = randomArray(3, false);
+        vint4Bytes = randomArray(4, false);
+        vint5Bytes = randomArray(5, false);
+        vintNegBytes = randomArray(random.nextInt(1, 6), true);
+    }
+
+    private void writeLoop(int[] nums) throws IOException {
+        for (int reps = 0; reps < WRITES_PER_ITERATION / nums.length; reps++) {
+            for (var n : nums) {
+                output.writeVInt(n);
+            }
+        }
+        output.seek(0);
+    }
+
+    @Benchmark
+    @OperationsPerInvocation(WRITES_PER_ITERATION)
+    public void writeVInt1() throws IOException {
+        writeLoop(vint1Byte);
+    }
+
+    @Benchmark
+    @OperationsPerInvocation(WRITES_PER_ITERATION)
+    public void writeVInt2() throws IOException {
+        writeLoop(vint2Bytes);
+    }
+
+    @Benchmark
+    @OperationsPerInvocation(WRITES_PER_ITERATION)
+    public void writeVInt3() throws IOException {
+        writeLoop(vint3Bytes);
+    }
+
+    @Benchmark
+    @OperationsPerInvocation(WRITES_PER_ITERATION)
+    public void writeVInt4() throws IOException {
+        writeLoop(vint4Bytes);
+    }
+
+    @Benchmark
+    @OperationsPerInvocation(WRITES_PER_ITERATION)
+    public void writeVInt5() throws IOException {
+        writeLoop(vint5Bytes);
+    }
+
+    @Benchmark
+    @OperationsPerInvocation(WRITES_PER_ITERATION)
+    public void writeVIntNeg() throws IOException {
+        writeLoop(vintNegBytes);
+    }
+
+    // recycle same page, we never read previous pages
+    private static class SinglePageStream implements Recycler<BytesRef> {
+
+        private final BytesRef page = new BytesRef(new byte[PAGE_SIZE], 0, PAGE_SIZE);
+
+        @Override
+        public V<BytesRef> obtain() {
+            return new V<>() {
+                @Override
+                public BytesRef v() {
+                    return page;
+                }
+
+                @Override
+                public boolean isRecycled() {
+                    return true;
+                }
+
+                @Override
+                public void close() {}
+            };
+        }
+
+        @Override
+        public int pageSize() {
+            return PAGE_SIZE;
+        }
+    }
+}

benchmarks/src/main/java/org/elasticsearch/benchmark/compute/operator/AggregatorBenchmark.java

@@ -733,6 +733,6 @@ private static void run(String grouping, String op, String blockType, String fil
     }
 
     static DriverContext driverContext() {
-        return new DriverContext(BigArrays.NON_RECYCLING_INSTANCE, blockFactory);
+        return new DriverContext(BigArrays.NON_RECYCLING_INSTANCE, blockFactory, null);
     }
 }