Merge branch 'main' into esql-top-n-intermediate-error

# Conflicts:
#	x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/aggregation/CountGroupingAggregatorFunction.java

Author: ivancea

docs/changelog/130924.yaml

@@ -0,0 +1,6 @@
+pr: 130924
+summary: Check field data type before casting when applying geo distance sort
+area: Search
+type: bug
+issues:
+ - 129500

docs/reference/elasticsearch/index.md

@@ -1,4 +1,4 @@
-# Elasticsearch and index management
+# Elasticsearch
 
 This section contains reference information for {{es}} and index management features.
 
@@ -7,7 +7,7 @@ To learn more about {{es}} features and how to get started, refer to the [{{es}}
 For more details about query and scripting languages, check these sections:
 * [Query languages](../query-languages/index.md)
 * [Scripting languages](../scripting-languages/index.md)
- 
+
 {{es}} also provides the following REST APIs:
 
 * [{{es}} API](https://www.elastic.co/docs/api/doc/elasticsearch)

docs/reference/elasticsearch/rest-apis/index.md

@@ -564,18 +564,18 @@ tests:
 - class: org.elasticsearch.cluster.ClusterStateSerializationTests
   method: testSerializationPreMultiProject
   issue: https://github.com/elastic/elasticsearch/issues/130872
-- class: org.elasticsearch.cluster.coordination.votingonly.VotingOnlyNodePluginTests
-  method: testPreferFullMasterOverVotingOnlyNodes
-  issue: https://github.com/elastic/elasticsearch/issues/130883
 - class: org.elasticsearch.search.SearchWithRejectionsIT
   method: testOpenContextsAfterRejections
   issue: https://github.com/elastic/elasticsearch/issues/130821
-- class: org.elasticsearch.cluster.coordination.votingonly.VotingOnlyNodePluginTests
-  method: testVotingOnlyNodesCannotBeMasterWithoutFullMasterNodes
-  issue: https://github.com/elastic/elasticsearch/issues/130979
 - class: org.elasticsearch.packaging.test.DockerTests
   method: test082CannotUseEnvVarsAndFiles
   issue: https://github.com/elastic/elasticsearch/issues/129808
+- class: org.elasticsearch.xpack.esql.action.EsqlActionBreakerIT
+  method: testExtractFields
+  issue: https://github.com/elastic/elasticsearch/issues/130501
+- class: org.elasticsearch.xpack.esql.action.EsqlActionBreakerIT
+  method: testRowStatsProjectGroupByInt
+  issue: https://github.com/elastic/elasticsearch/issues/131024
 
 # Examples:
 #

muted-tests.yml

@@ -189,37 +189,46 @@ public StoredContext newEmptySystemContext() {
      * moving tracing-related fields to different names so that a new child span can be started. This child span will pick up
      * the moved fields and use them to establish the parent-child relationship.
      *
+     * Response headers will be propagated. If no parent span is in progress (meaning there's no trace context), this will behave exactly
+     * the same way as {@link #newStoredContextPreservingResponseHeaders}.
+     *
      * @return a stored context, which can be restored when this context is no longer needed.
      */
     public StoredContext newTraceContext() {
         final ThreadContextStruct originalContext = threadLocal.get();
-        final Map<String, String> newRequestHeaders = new HashMap<>(originalContext.requestHeaders);
-        final Map<String, Object> newTransientHeaders = new HashMap<>(originalContext.transientHeaders);
 
-        final String previousTraceParent = newRequestHeaders.remove(Task.TRACE_PARENT_HTTP_HEADER);
-        if (previousTraceParent != null) {
-            newTransientHeaders.put(Task.PARENT_TRACE_PARENT_HEADER, previousTraceParent);
-        }
+        // this is the context when this method returns
+        final ThreadContextStruct newContext;
+        if (originalContext.hasTraceContext() == false) {
+            newContext = originalContext;
+        } else {
+            final Map<String, String> newRequestHeaders = new HashMap<>(originalContext.requestHeaders);
+            final Map<String, Object> newTransientHeaders = new HashMap<>(originalContext.transientHeaders);
 
-        final String previousTraceState = newRequestHeaders.remove(Task.TRACE_STATE);
-        if (previousTraceState != null) {
-            newTransientHeaders.put(Task.PARENT_TRACE_STATE, previousTraceState);
-        }
+            final String previousTraceParent = newRequestHeaders.remove(Task.TRACE_PARENT_HTTP_HEADER);
+            if (previousTraceParent != null) {
+                newTransientHeaders.put(Task.PARENT_TRACE_PARENT_HEADER, previousTraceParent);
+            }
 
-        final Object previousTraceContext = newTransientHeaders.remove(Task.APM_TRACE_CONTEXT);
-        if (previousTraceContext != null) {
-            newTransientHeaders.put(Task.PARENT_APM_TRACE_CONTEXT, previousTraceContext);
-        }
+            final String previousTraceState = newRequestHeaders.remove(Task.TRACE_STATE);
+            if (previousTraceState != null) {
+                newTransientHeaders.put(Task.PARENT_TRACE_STATE, previousTraceState);
+            }
 
-        // this is the context when this method returns
-        final ThreadContextStruct newContext = new ThreadContextStruct(
-            newRequestHeaders,
-            originalContext.responseHeaders,
-            newTransientHeaders,
-            originalContext.isSystemContext,
-            originalContext.warningHeadersSize
-        );
-        threadLocal.set(newContext);
+            final Object previousTraceContext = newTransientHeaders.remove(Task.APM_TRACE_CONTEXT);
+            if (previousTraceContext != null) {
+                newTransientHeaders.put(Task.PARENT_APM_TRACE_CONTEXT, previousTraceContext);
+            }
+
+            newContext = new ThreadContextStruct(
+                newRequestHeaders,
+                originalContext.responseHeaders,
+                newTransientHeaders,
+                originalContext.isSystemContext,
+                originalContext.warningHeadersSize
+            );
+            threadLocal.set(newContext);
+        }
         // Tracing shouldn't interrupt the propagation of response headers, so in the same as
         // #newStoredContextPreservingResponseHeaders(), pass on any potential changes to the response headers.
         return () -> {
@@ -233,10 +242,11 @@ public StoredContext newTraceContext() {
     }
 
     public boolean hasTraceContext() {
-        final ThreadContextStruct context = threadLocal.get();
-        return context.requestHeaders.containsKey(Task.TRACE_PARENT_HTTP_HEADER)
-            || context.requestHeaders.containsKey(Task.TRACE_STATE)
-            || context.transientHeaders.containsKey(Task.APM_TRACE_CONTEXT);
+        return threadLocal.get().hasTraceContext();
+    }
+
+    public boolean hasParentTraceContext() {
+        return threadLocal.get().hasParentTraceContext();
     }
 
     /**
@@ -853,6 +863,18 @@ private ThreadContextStruct putResponseHeaders(Map<String, Set<String>> headers)
             return new ThreadContextStruct(requestHeaders, newResponseHeaders, transientHeaders, isSystemContext);
         }
 
+        private boolean hasTraceContext() {
+            return requestHeaders.containsKey(Task.TRACE_PARENT_HTTP_HEADER)
+                || requestHeaders.containsKey(Task.TRACE_STATE)
+                || transientHeaders.containsKey(Task.APM_TRACE_CONTEXT);
+        }
+
+        private boolean hasParentTraceContext() {
+            return transientHeaders.containsKey(Task.PARENT_TRACE_PARENT_HEADER)
+                || transientHeaders.containsKey(Task.PARENT_TRACE_STATE)
+                || transientHeaders.containsKey(Task.PARENT_APM_TRACE_CONTEXT);
+        }
+
         private void logWarningHeaderThresholdExceeded(long threshold, Setting<?> thresholdSetting) {
             // If available, log some selected headers to help identifying the source of the request.
             // Note: Only Task.HEADERS_TO_COPY are guaranteed to be preserved at this point.

server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java

@@ -599,7 +599,13 @@ private IndexGeoPointFieldData fieldData(SearchExecutionContext context) {
                 throw new IllegalArgumentException("failed to find mapper for [" + fieldName + "] for geo distance based sort");
             }
         }
-        return context.getForField(fieldType, MappedFieldType.FielddataOperation.SEARCH);
+        IndexFieldData<?> indexFieldData = context.getForField(fieldType, MappedFieldType.FielddataOperation.SEARCH);
+        if (indexFieldData instanceof IndexGeoPointFieldData) {
+            return (IndexGeoPointFieldData) indexFieldData;
+        }
+        throw new IllegalArgumentException(
+            "unable to apply geo distance sort to field [" + fieldName + "] of type [" + fieldType.typeName() + "]"
+        );
     }
 
     private Nested nested(SearchExecutionContext context) throws IOException {

server/src/main/java/org/elasticsearch/search/sort/GeoDistanceSortBuilder.java

@@ -44,6 +44,7 @@ public class Task implements Traceable {
      * TRACE_PARENT once parsed in RestController.tryAllHandler is not preserved
      * has to be declared as a header copied over from http request.
      * May also be used internally when APM is enabled.
+     * https://www.w3.org/TR/trace-context-1/#traceparent-header
      */
     public static final String TRACE_PARENT_HTTP_HEADER = "traceparent";
 
@@ -53,6 +54,10 @@ public class Task implements Traceable {
      */
     public static final String TRACE_ID = "trace.id";
 
+    /**
+     * Optional request header carrying vendor-specific trace information.
+     * https://www.w3.org/TR/trace-context-1/#tracestate-header
+     */
     public static final String TRACE_STATE = "tracestate";
 
     public static final String TRACE_START_TIME = "trace.starttime";

server/src/main/java/org/elasticsearch/tasks/Task.java

@@ -179,12 +179,9 @@ public Task register(String type, String action, TaskAwareRequest request, boole
     // package private for testing
     void startTrace(ThreadContext threadContext, Task task) {
         TaskId parentTask = task.getParentTaskId();
-        Map<String, Object> attributes = Map.of(
-            Tracer.AttributeKeys.TASK_ID,
-            task.getId(),
-            Tracer.AttributeKeys.PARENT_TASK_ID,
-            parentTask.toString()
-        );
+        Map<String, Object> attributes = parentTask.isSet()
+            ? Map.of(Tracer.AttributeKeys.TASK_ID, task.getId(), Tracer.AttributeKeys.PARENT_TASK_ID, parentTask.toString())
+            : Map.of(Tracer.AttributeKeys.TASK_ID, task.getId());
         tracer.startTrace(threadContext, task, task.getAction(), attributes);
     }
 

server/src/main/java/org/elasticsearch/tasks/TaskManager.java

@@ -34,12 +34,14 @@
 
 import static com.carrotsearch.randomizedtesting.RandomizedTest.randomAsciiLettersOfLengthBetween;
 import static org.elasticsearch.tasks.Task.HEADERS_TO_COPY;
+import static org.hamcrest.Matchers.anEmptyMap;
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasItem;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.nullValue;
 import static org.hamcrest.Matchers.sameInstance;
@@ -1161,6 +1163,75 @@ public void testNewEmptySystemContext() {
         assertNotNull(threadContext.getHeader(header));
     }
 
+    public void testNewTraceContext() {
+        final var threadContext = new ThreadContext(Settings.EMPTY);
+
+        var rootTraceContext = Map.of(Task.TRACE_PARENT_HTTP_HEADER, randomIdentifier(), Task.TRACE_STATE, randomIdentifier());
+        var apmTraceContext = new Object();
+        var responseKey = randomIdentifier();
+        var responseValue = randomAlphaOfLength(10);
+
+        threadContext.putHeader(rootTraceContext);
+        threadContext.putTransient(Task.APM_TRACE_CONTEXT, apmTraceContext);
+
+        assertThat(threadContext.hasTraceContext(), equalTo(true));
+        assertThat(threadContext.hasParentTraceContext(), equalTo(false));
+
+        try (var ignored = threadContext.newTraceContext()) {
+            assertThat(threadContext.hasTraceContext(), equalTo(false)); // no trace started yet
+            assertThat(threadContext.hasParentTraceContext(), equalTo(true));
+
+            assertThat(threadContext.getHeaders(), is(anEmptyMap()));
+            assertThat(
+                threadContext.getTransientHeaders(),
+                equalTo(
+                    Map.of(
+                        Task.PARENT_TRACE_PARENT_HEADER,
+                        rootTraceContext.get(Task.TRACE_PARENT_HTTP_HEADER),
+                        Task.PARENT_TRACE_STATE,
+                        rootTraceContext.get(Task.TRACE_STATE),
+                        Task.PARENT_APM_TRACE_CONTEXT,
+                        apmTraceContext
+                    )
+                )
+            );
+            // response headers shall be propagated
+            threadContext.addResponseHeader(responseKey, responseValue);
+        }
+
+        assertThat(threadContext.hasTraceContext(), equalTo(true));
+        assertThat(threadContext.hasParentTraceContext(), equalTo(false));
+
+        assertThat(threadContext.getHeaders(), equalTo(rootTraceContext));
+        assertThat(threadContext.getTransientHeaders(), equalTo(Map.of(Task.APM_TRACE_CONTEXT, apmTraceContext)));
+        assertThat(threadContext.getResponseHeaders(), equalTo(Map.of(responseKey, List.of(responseValue))));
+    }
+
+    public void testNewTraceContextWithoutParentTrace() {
+        final var threadContext = new ThreadContext(Settings.EMPTY);
+
+        var responseKey = randomIdentifier();
+        var responseValue = randomAlphaOfLength(10);
+
+        assertThat(threadContext.hasTraceContext(), equalTo(false));
+        assertThat(threadContext.hasParentTraceContext(), equalTo(false));
+
+        try (var ignored = threadContext.newTraceContext()) {
+            assertTrue(threadContext.isDefaultContext());
+            assertThat(threadContext.hasTraceContext(), equalTo(false));
+            assertThat(threadContext.hasParentTraceContext(), equalTo(false));
+
+            // discared, just making sure the context is isolated
+            threadContext.putTransient(randomIdentifier(), randomAlphaOfLength(10));
+            // response headers shall be propagated
+            threadContext.addResponseHeader(responseKey, responseValue);
+        }
+
+        assertThat(threadContext.getHeaders(), is(anEmptyMap()));
+        assertThat(threadContext.getTransientHeaders(), is(anEmptyMap()));
+        assertThat(threadContext.getResponseHeaders(), equalTo(Map.of(responseKey, List.of(responseValue))));
+    }
+
     public void testRestoreExistingContext() {
         final var threadContext = new ThreadContext(Settings.EMPTY);
         final var header = randomIdentifier();

server/src/test/java/org/elasticsearch/common/util/concurrent/ThreadContextTests.java

@@ -23,6 +23,7 @@
 import org.elasticsearch.index.mapper.GeoPointFieldMapper;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.NestedPathFieldMapper;
+import org.elasticsearch.index.mapper.NumberFieldMapper;
 import org.elasticsearch.index.query.GeoValidationMethod;
 import org.elasticsearch.index.query.MatchAllQueryBuilder;
 import org.elasticsearch.index.query.MatchNoneQueryBuilder;
@@ -99,6 +100,9 @@ public static GeoDistanceSortBuilder randomGeoDistanceSortBuilder() {
 
     @Override
     protected MappedFieldType provideMappedFieldType(String name) {
+        if (name.equals("double")) {
+            return new NumberFieldMapper.NumberFieldType(name, NumberFieldMapper.NumberType.DOUBLE);
+        }
         return new GeoPointFieldMapper.GeoPointFieldType(name);
     }
 
@@ -531,6 +535,12 @@ public void testBuildInvalidPoints() throws IOException {
             );
             assertEquals("illegal longitude value [-360.0] for [GeoDistanceSort] for field [fieldName].", ex.getMessage());
         }
+        {
+            GeoDistanceSortBuilder sortBuilder = new GeoDistanceSortBuilder("double", 0.0, 180.0);
+            sortBuilder.validation(GeoValidationMethod.STRICT);
+            IllegalArgumentException ex = expectThrows(IllegalArgumentException.class, () -> sortBuilder.build(searchExecutionContext));
+            assertEquals("unable to apply geo distance sort to field [double] of type [double]", ex.getMessage());
+        }
     }
 
     /**