Merge branch 'main' into layout-build-npe

Author: fang-xing-esql

docs/changelog/124913.yaml

@@ -0,0 +1,5 @@
+pr: 124913
+summary: Report `original_types`
+area: ES|QL
+type: enhancement
+issues: []

docs/internal/DistributedArchitectureGuide.md

@@ -110,6 +110,16 @@ to communicate with Elasticsearch.
 
 ### Cluster State
 
+[ClusterState]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/cluster/ClusterState.java
+[Metadata]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java
+[ProjectMetadata]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/cluster/metadata/ProjectMetadata.java
+
+The [Metadata] of a [ClusterState] is persisted on disk and comprises information from two categories:
+1. Cluster scope information such as `clusterUUID`, `CoordinationMetadata`
+2. Project scope information ([ProjectMetadata]) such as indices and templates belong to each project.
+
+Some concepts are applicable to both cluster and project scopes, e.g. [persistent tasks](#persistent-tasks). The state of a persistent task is therefore stored accordingly depending on the task's scope.
+
 #### Master Service
 
 #### Cluster State Publication
@@ -306,7 +316,7 @@ policies.
 
 ### How cluster capacity is determined
 
-[AutoscalingMetadata][] implements [Metadata.Custom][] in order to persist autoscaling policies. Each
+[AutoscalingMetadata][] implements [Metadata.ClusterCustom][] in order to persist autoscaling policies. Each
 Decider is an implementation of [AutoscalingDeciderService][]. The [AutoscalingCalculateCapacityService][]
 is responsible for running the calculation.
 
@@ -322,7 +332,7 @@ calls [through the CapacityResponseCache][], into the `AutoscalingCalculateCapac
 concurrent callers.
 
 [AutoscalingMetadata]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/AutoscalingMetadata.java#L38
-[Metadata.Custom]: https://github.com/elastic/elasticsearch/blob/v8.13.2/server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java#L141-L145
+[Metadata.ClusterCustom]: https://github.com/elastic/elasticsearch/blob/f461731a30a6fe55d7d7b343d38426ddca1ac873/server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java#L147
 [AutoscalingDeciderService]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/capacity/AutoscalingDeciderService.java#L16-L19
 [AutoscalingCalculateCapacityService]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/capacity/AutoscalingCalculateCapacityService.java#L43
 
@@ -505,6 +515,8 @@ The [Task management API] also exposes an endpoint where a task ID can be specif
 
 [PersistentTaskPlugin]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/plugins/PersistentTaskPlugin.java
 [PersistentTasksExecutor]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/persistent/PersistentTasksExecutor.java
+[PersistentTasksExecutor.Scope.Cluster]:https://github.com/elastic/elasticsearch/blob/f461731a30a6fe55d7d7b343d38426ddca1ac873/server/src/main/java/org/elasticsearch/persistent/PersistentTasksExecutor.java#L52
+[PersistentTasksExecutor.Scope.Project]:https://github.com/elastic/elasticsearch/blob/f461731a30a6fe55d7d7b343d38426ddca1ac873/server/src/main/java/org/elasticsearch/persistent/PersistentTasksExecutor.java#L48
 [PersistentTasksExecutorRegistry]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/persistent/PersistentTasksExecutorRegistry.java
 [PersistentTasksNodeService]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/persistent/PersistentTasksNodeService.java
 [PersistentTasksClusterService]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/persistent/PersistentTasksClusterService.java
@@ -513,13 +525,14 @@ The [Task management API] also exposes an endpoint where a task ID can be specif
 [HealthNodeTaskExecutor]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/health/node/selection/HealthNodeTaskExecutor.java
 [SystemIndexMigrationExecutor]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/upgrades/SystemIndexMigrationExecutor.java
 [PersistentTasksCustomMetadata]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/persistent/PersistentTasksCustomMetadata.java
+[ClusterPersistentTasksCustomMetadata]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/persistent/ClusterPersistentTasksCustomMetadata.java
 [PersistentTasksCustomMetadata.PersistentTask]:https://github.com/elastic/elasticsearch/blob/d466ad1c3c4cedc7d5f6ab5794abe7bfd72aef4e/server/src/main/java/org/elasticsearch/persistent/PersistentTasksCustomMetadata.java#L305
 
 Up until now we have discussed only ephemeral tasks. If we want a task to survive node failures, it needs to be registered as a persistent task at the cluster level.
 
-Plugins can register persistent tasks definitions by implementing [PersistentTaskPlugin] and returning one or more [PersistentTasksExecutor] instances. These are collated into a [PersistentTasksExecutorRegistry] which is provided to [PersistentTasksNodeService] active on each node in the cluster, and a [PersistentTasksClusterService] active on the master.
+Plugins can register persistent tasks definitions by implementing [PersistentTaskPlugin] and returning one or more [PersistentTasksExecutor] instances. These are collated into a [PersistentTasksExecutorRegistry] which is provided to [PersistentTasksNodeService] active on each node in the cluster, and a [PersistentTasksClusterService] active on the master. A [PersistentTasksExecutor] can declare either [project][PersistentTasksExecutor.Scope.Project] or [cluster][PersistentTasksExecutor.Scope.Cluster] scope, but not both. A project scope task is not able to access data on a different project.
 
-The [PersistentTasksClusterService] runs on the master to manage the set of running persistent tasks. It periodically checks that all persistent tasks are assigned to live nodes and handles the creation, completion, removal and updates-to-the-state of persistent task instances in the cluster state (see [PersistentTasksCustomMetadata]).
+The [PersistentTasksClusterService] runs on the master to manage the set of running persistent tasks. It periodically checks that all persistent tasks are assigned to live nodes and handles the creation, completion, removal and updates-to-the-state of persistent task instances in the cluster state (see [PersistentTasksCustomMetadata] and [ClusterPersistentTasksCustomMetadata]).
 
 The [PersistentTasksNodeService] monitors the cluster state to:
  - Start any tasks allocated to it (tracked in the local [TaskManager] by an [AllocatedPersistentTask])
@@ -529,7 +542,7 @@ If a node leaves the cluster while it has a persistent task allocated to it, the
 
 Some examples of the use of persistent tasks include:
  - [ShardFollowTasksExecutor]: Defined by [cross-cluster replication](#cross-cluster-replication-ccr) to poll a remote cluster for updates
- - [HealthNodeTaskExecutor]: Used to schedule work related to monitoring cluster health
+ - [HealthNodeTaskExecutor]: Used to schedule work related to monitoring cluster health. This is currently the only example of a cluster scope persistent task.
  - [SystemIndexMigrationExecutor]: Manages the migration of system indices after an upgrade
 
 ### Integration with APM

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

@@ -203,6 +203,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion ESQL_AGGREGATE_METRIC_DOUBLE_LITERAL = def(9_035_0_00);
     public static final TransportVersion INDEX_METADATA_INCLUDES_RECENT_WRITE_LOAD = def(9_036_0_00);
     public static final TransportVersion RERANK_COMMON_OPTIONS_ADDED = def(9_037_0_00);
+    public static final TransportVersion ESQL_REPORT_ORIGINAL_TYPES = def(9_038_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/Attribute.java

@@ -137,4 +137,14 @@ public String nodeString() {
     }
 
     protected abstract String label();
+
+    /**
+     * If this field is unsupported this contains the underlying ES types. If there
+     * is a type conflict this will have many elements, some or all of which may
+     * be actually supported types.
+     */
+    @Nullable
+    public List<String> originalTypes() {
+        return null;
+    }
 }

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/type/UnsupportedEsField.java

@@ -6,10 +6,14 @@
  */
 package org.elasticsearch.xpack.esql.core.type;
 
+import org.elasticsearch.TransportVersions;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.xpack.esql.core.util.PlanStreamInput;
+import org.elasticsearch.xpack.esql.core.util.PlanStreamOutput;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.TreeMap;
@@ -23,32 +27,39 @@
  */
 public class UnsupportedEsField extends EsField {
 
-    private final String originalType;
+    private final List<String> originalTypes;
     private final String inherited; // for fields belonging to parents (or grandparents) that have an unsupported type
 
-    public UnsupportedEsField(String name, String originalType) {
-        this(name, originalType, null, new TreeMap<>());
+    public UnsupportedEsField(String name, List<String> originalTypes) {
+        this(name, originalTypes, null, new TreeMap<>());
     }
 
-    public UnsupportedEsField(String name, String originalType, String inherited, Map<String, EsField> properties) {
+    public UnsupportedEsField(String name, List<String> originalTypes, String inherited, Map<String, EsField> properties) {
         super(name, DataType.UNSUPPORTED, properties, false);
-        this.originalType = originalType;
+        this.originalTypes = originalTypes;
         this.inherited = inherited;
     }
 
     public UnsupportedEsField(StreamInput in) throws IOException {
-        this(
-            readCachedStringWithVersionCheck(in),
-            readCachedStringWithVersionCheck(in),
-            in.readOptionalString(),
-            in.readImmutableMap(EsField::readFrom)
-        );
+        this(readCachedStringWithVersionCheck(in), readOriginalTypes(in), in.readOptionalString(), in.readImmutableMap(EsField::readFrom));
+    }
+
+    private static List<String> readOriginalTypes(StreamInput in) throws IOException {
+        if (in.getTransportVersion().onOrAfter(TransportVersions.ESQL_REPORT_ORIGINAL_TYPES)) {
+            return in.readCollectionAsList(i -> ((PlanStreamInput) i).readCachedString());
+        } else {
+            return List.of(readCachedStringWithVersionCheck(in).split(","));
+        }
     }
 
     @Override
     public void writeContent(StreamOutput out) throws IOException {
         writeCachedStringWithVersionCheck(out, getName());
-        writeCachedStringWithVersionCheck(out, getOriginalType());
+        if (out.getTransportVersion().onOrAfter(TransportVersions.ESQL_REPORT_ORIGINAL_TYPES)) {
+            out.writeCollection(getOriginalTypes(), (o, s) -> ((PlanStreamOutput) o).writeCachedString(s));
+        } else {
+            writeCachedStringWithVersionCheck(out, String.join(",", getOriginalTypes()));
+        }
         out.writeOptionalString(getInherited());
         out.writeMap(getProperties(), (o, x) -> x.writeTo(out));
     }
@@ -57,8 +68,8 @@ public String getWriteableName() {
         return "UnsupportedEsField";
     }
 
-    public String getOriginalType() {
-        return originalType;
+    public List<String> getOriginalTypes() {
+        return originalTypes;
     }
 
     public String getInherited() {
@@ -81,11 +92,11 @@ public boolean equals(Object o) {
             return false;
         }
         UnsupportedEsField that = (UnsupportedEsField) o;
-        return Objects.equals(originalType, that.originalType) && Objects.equals(inherited, that.inherited);
+        return Objects.equals(originalTypes, that.originalTypes) && Objects.equals(inherited, that.inherited);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(super.hashCode(), originalType, inherited);
+        return Objects.hash(super.hashCode(), originalTypes, inherited);
     }
 }