Merge branch 'main' into entitlements/files-policy-base-dir

Author: ldematte

docs/changelog/122074.yaml

@@ -0,0 +1,8 @@
+pr: 122074
+summary: If the Transform is configured to write to an alias as its destination index,
+  when the delete_dest_index parameter is set to true, then the Delete API will now
+  delete the write index backing the alias
+area: Transform
+type: bug
+issues:
+ - 121913

docs/changelog/122365.yaml

@@ -0,0 +1,5 @@
+pr: 122365
+summary: Fix handling of auto expand replicas for stateless indices
+area: "Search"
+type: bug
+issues: []

muted-tests.yml

@@ -320,9 +320,6 @@ tests:
 - class: org.elasticsearch.xpack.security.profile.ProfileIntegTests
   method: testGetUsersWithProfileUid
   issue: https://github.com/elastic/elasticsearch/issues/121483
-- class: org.elasticsearch.xpack.transform.checkpoint.TransformCCSCanMatchIT
-  method: testTransformLifecycle_RangeQueryThatMatchesNoShards
-  issue: https://github.com/elastic/elasticsearch/issues/121480
 - class: org.elasticsearch.xpack.security.profile.ProfileIntegTests
   method: testSuggestProfilesWithHint
   issue: https://github.com/elastic/elasticsearch/issues/121116
@@ -404,6 +401,9 @@ tests:
   issue: https://github.com/elastic/elasticsearch/issues/122377
 - class: org.elasticsearch.repositories.blobstore.testkit.analyze.HdfsRepositoryAnalysisRestIT
   issue: https://github.com/elastic/elasticsearch/issues/122378
+- class: org.elasticsearch.xpack.esql.action.EsqlActionBreakerIT
+  method: testStatsMissingFieldWithStats
+  issue: https://github.com/elastic/elasticsearch/issues/122327
 
 # Examples:
 #

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -185,6 +185,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion ESQL_DRIVER_TASK_DESCRIPTION = def(9_005_0_00);
     public static final TransportVersion ESQL_RETRY_ON_SHARD_LEVEL_FAILURE = def(9_006_0_00);
     public static final TransportVersion ESQL_PROFILE_ASYNC_NANOS = def(9_007_00_0);
+    public static final TransportVersion ESQL_LOOKUP_JOIN_SOURCE_TEXT = def(9_008_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,
@@ -197,6 +198,8 @@ static TransportVersion def(int id) {
      * A new transport version should be added EVERY TIME a change is made to the serialization protocol of one or more classes. Each
      * transport version should only be used in a single merged commit (apart from the BwC versions copied from o.e.Version, ≤V_8_8_1).
      *
+     * More information about versions and backporting at docs/internal/Versioning.md
+     *
      * ADDING A TRANSPORT VERSION
      * To add a new transport version, add a new constant at the bottom of the list, above this comment. Don't add other lines,
      * comments, etc. The version id has the following layout:

server/src/main/java/org/elasticsearch/cluster/metadata/AutoExpandReplicas.java

@@ -156,9 +156,8 @@ public static Map<Integer, List<String>> getAutoExpandReplicaChanges(
                 )) {
                     if (indexMetadata.getNumberOfReplicas() == 0) {
                         nrReplicasChanged.computeIfAbsent(1, ArrayList::new).add(indexMetadata.getIndex().getName());
-                    } else {
-                        continue;
                     }
+                    continue;
                 }
                 if (allocation == null) {
                     allocation = allocationSupplier.get();

server/src/main/java/org/elasticsearch/search/aggregations/InternalOrder.java

@@ -438,16 +438,7 @@ public static boolean isKeyDesc(BucketOrder order) {
      * @return {@code true} if the order matches, {@code false} otherwise.
      */
     private static boolean isOrder(BucketOrder order, BucketOrder expected) {
-        if (order == expected) {
-            return true;
-        } else if (order instanceof CompoundOrder) {
-            // check if its a compound order with the first element that matches
-            List<BucketOrder> orders = ((CompoundOrder) order).orderElements;
-            if (orders.size() >= 1) {
-                return isOrder(orders.get(0), expected);
-            }
-        }
-        return false;
+        return order == expected || (order instanceof CompoundOrder compoundOrder && compoundOrder.orderElements.getFirst() == expected);
     }
 
     /**

server/src/test/java/org/elasticsearch/cluster/metadata/AutoExpandReplicasTests.java

@@ -20,8 +20,10 @@
 import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
 import org.elasticsearch.cluster.routing.RoutingNodesHelper;
 import org.elasticsearch.cluster.routing.ShardRoutingState;
+import org.elasticsearch.cluster.routing.allocation.ExistingShardsAllocator;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Strings;
+import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.indices.cluster.ClusterStateChanges;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.threadpool.TestThreadPool;
@@ -31,11 +33,14 @@
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
+import static org.elasticsearch.cluster.metadata.IndexMetadata.INDEX_AUTO_EXPAND_REPLICAS_SETTING;
 import static org.elasticsearch.cluster.metadata.IndexMetadata.SETTING_AUTO_EXPAND_REPLICAS;
+import static org.elasticsearch.cluster.metadata.IndexMetadata.SETTING_NUMBER_OF_REPLICAS;
 import static org.elasticsearch.cluster.metadata.IndexMetadata.SETTING_NUMBER_OF_SHARDS;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.everyItem;
@@ -221,4 +226,48 @@ public void testCalculateDesiredNumberOfReplicas() {
         assertThat(autoExpandReplicas.calculateDesiredNumberOfReplicas(matchingNodes), equalTo(Math.max(lowerBound, matchingNodes - 1)));
         assertThat(autoExpandReplicas.calculateDesiredNumberOfReplicas(max + 1), equalTo(max));
     }
+
+    public void testGetAutoExpandReplicaChangesStatelessIndices() {
+        {
+            // number of replicas is adjusted to 1 when it is initialized to 0
+            Metadata metadata = Metadata.builder()
+                .put(
+                    IndexMetadata.builder("test")
+                        .settings(
+                            Settings.builder()
+                                .put(ExistingShardsAllocator.EXISTING_SHARDS_ALLOCATOR_SETTING.getKey(), "stateless")
+                                .put("index.version.created", IndexVersion.current())
+                                .put(SETTING_NUMBER_OF_SHARDS, 1)
+                                .put(SETTING_NUMBER_OF_REPLICAS, 0)
+                                .put(INDEX_AUTO_EXPAND_REPLICAS_SETTING.getKey(), "0-all")
+                        )
+                )
+                .build();
+            Map<Integer, List<String>> autoExpandReplicaChanges = AutoExpandReplicas.getAutoExpandReplicaChanges(metadata, null);
+            assertEquals(1, autoExpandReplicaChanges.size());
+            List<String> indices = autoExpandReplicaChanges.get(1);
+            assertEquals(1, indices.size());
+            assertEquals("test", indices.getFirst());
+        }
+        {
+            // no changes when number of replicas is set to anything other than 0
+            Metadata metadata = Metadata.builder()
+                .put(
+                    IndexMetadata.builder("test")
+                        .settings(
+                            Settings.builder()
+                                .put(ExistingShardsAllocator.EXISTING_SHARDS_ALLOCATOR_SETTING.getKey(), "stateless")
+                                .put("index.version.created", IndexVersion.current())
+                                .put(SETTING_NUMBER_OF_SHARDS, 1)
+                                .put(SETTING_NUMBER_OF_REPLICAS, randomIntBetween(1, 10))
+                                .put(INDEX_AUTO_EXPAND_REPLICAS_SETTING.getKey(), "0-all")
+                        )
+                )
+                .build();
+            Map<Integer, List<String>> autoExpandReplicaChanges = AutoExpandReplicas.getAutoExpandReplicaChanges(metadata, () -> {
+                throw new UnsupportedOperationException();
+            });
+            assertEquals(0, autoExpandReplicaChanges.size());
+        }
+    }
 }

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/AsyncOperator.java

@@ -17,6 +17,7 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.seqno.LocalCheckpointTracker;
@@ -34,6 +35,11 @@
  * to reduce communication overhead and fetches a {@code Fetched} at a time.
  * It's the responsibility of subclasses to transform that {@code Fetched} into
  * output.
+ * <p>
+ *     This operator will also take care of merging response headers from the thread context into the main thread,
+ *     which <b>must</b> be the one that closes this.
+ * </p>
+ *
  * @see #performAsync(Page, ActionListener)
  */
 public abstract class AsyncOperator<Fetched> implements Operator {
@@ -45,6 +51,7 @@ public abstract class AsyncOperator<Fetched> implements Operator {
     private final DriverContext driverContext;
 
     private final int maxOutstandingRequests;
+    private final ResponseHeadersCollector responseHeadersCollector;
     private final LongAdder processNanos = new LongAdder();
 
     private boolean finished = false;
@@ -66,9 +73,10 @@ public abstract class AsyncOperator<Fetched> implements Operator {
      *
      * @param maxOutstandingRequests the maximum number of outstanding requests
      */
-    public AsyncOperator(DriverContext driverContext, int maxOutstandingRequests) {
+    public AsyncOperator(DriverContext driverContext, ThreadContext threadContext, int maxOutstandingRequests) {
         this.driverContext = driverContext;
         this.maxOutstandingRequests = maxOutstandingRequests;
+        this.responseHeadersCollector = new ResponseHeadersCollector(threadContext);
     }
 
     @Override
@@ -97,6 +105,7 @@ public void addInput(Page input) {
             });
             final long startNanos = System.nanoTime();
             performAsync(input, ActionListener.runAfter(listener, () -> {
+                responseHeadersCollector.collect();
                 driverContext.removeAsyncAction();
                 processNanos.add(System.nanoTime() - startNanos);
             }));
@@ -172,6 +181,7 @@ public final void close() {
         finish();
         closed = true;
         discardResults();
+        responseHeadersCollector.finish();
         doClose();
     }
 

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/Warnings.java

@@ -102,12 +102,23 @@ private Warnings(int lineNumber, int columnNumber, String sourceText, String fir
     }
 
     public void registerException(Exception exception) {
+        registerException(exception.getClass(), exception.getMessage());
+    }
+
+    /**
+     * Register an exception to be included in the warnings.
+     * <p>
+     *     This overload avoids the need to instantiate the exception, which can be expensive.
+     *     Instead, it asks only the required pieces to build the warning.
+     * </p>
+     */
+    public void registerException(Class<? extends Exception> exceptionClass, String message) {
         if (addedWarnings < MAX_ADDED_WARNINGS) {
             if (addedWarnings == 0) {
                 addWarning(first);
             }
             // location needs to be added to the exception too, since the headers are deduplicated
-            addWarning(location + exception.getClass().getName() + ": " + exception.getMessage());
+            addWarning(location + exceptionClass.getName() + ": " + message);
             addedWarnings++;
         }
     }