Merge branch 'main' into feature/logsdb-ignore-dynamic-beyond-limit

Author: elasticmachine

docs/changelog/114951.yaml

@@ -0,0 +1,5 @@
+pr: 114951
+summary: Expose cluster-state role mappings in APIs
+area: Authentication
+type: bug
+issues: []

docs/changelog/115102.yaml

@@ -0,0 +1,6 @@
+pr: 115102
+summary: Watch Next Run Interval Resets On Shard Move or Node Restart
+area: Watcher
+type: bug
+issues:
+ - 111433

docs/changelog/115317.yaml

@@ -0,0 +1,5 @@
+pr: 115317
+summary: Revert "Add `ResolvedExpression` wrapper"
+area: Indices APIs
+type: bug
+issues: []

docs/changelog/115359.yaml

@@ -0,0 +1,6 @@
+pr: 115359
+summary: Adding support for simulate ingest mapping adddition for indices with mappings
+  that do not come from templates
+area: Ingest Node
+type: enhancement
+issues: []

docs/reference/esql/functions/kibana/definition/to_date_nanos.json

@@ -68,8 +68,16 @@ Synthetic source may sort `binary` values in order of their byte representation.
 ----
 PUT idx
 {
+  "settings": {
+    "index": {
+      "mapping": {
+        "source": {
+          "mode": "synthetic"
+        }
+      }
+    }
+  },
   "mappings": {
-    "_source": { "mode": "synthetic" },
     "properties": {
       "binary": { "type": "binary", "doc_values": true }
     }

docs/reference/mapping/types/binary.asciidoc

@@ -146,15 +146,18 @@ add, the more distributed the watches can be executed. If you add or remove
 replicas, all watches need to be reloaded. If a shard is relocated, the
 primary and all replicas of this particular shard will reload.
 
-Because the watches are executed on the node, where the watch shards are, you can create
-dedicated watcher nodes by using shard allocation filtering.
+Because the watches are executed on the node, where the watch shards are, you
+can create dedicated watcher nodes by using shard allocation filtering. To do this
+, configure nodes with a dedicated `node.attr.role: watcher` property.
 
-You could configure nodes with a dedicated `node.attr.role: watcher` property and
-then configure the `.watches` index like this:
+As the `.watches` index is a system index, you can't use the normal `.watcher/_settings`
+endpoint to modify its routing allocation. Instead, you can use the following dedicated
+endpoint to adjust the allocation of the `.watches` shards to the nodes with the
+`watcher` role attribute:
 
 [source,console]
 ------------------------
-PUT .watches/_settings
+PUT _watcher/settings
 {
   "index.routing.allocation.include.role": "watcher"
 }

docs/reference/watcher/how-watcher-works.asciidoc

@@ -43,13 +43,14 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.RemoteTransportException;
 
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.stream.Collectors;
 
 import static org.elasticsearch.ingest.geoip.GeoIpDownloader.DATABASES_INDEX;
 import static org.elasticsearch.ingest.geoip.GeoIpDownloader.GEOIP_DOWNLOADER;
@@ -238,14 +239,11 @@ public void clusterChanged(ClusterChangedEvent event) {
     }
 
     static boolean hasAtLeastOneGeoipProcessor(ClusterState clusterState) {
-        if (pipelineConfigurationsWithGeoIpProcessor(clusterState, true).isEmpty() == false) {
+        if (pipelinesWithGeoIpProcessor(clusterState, true).isEmpty() == false) {
             return true;
         }
 
-        Set<String> checkReferencedPipelines = pipelineConfigurationsWithGeoIpProcessor(clusterState, false).stream()
-            .map(PipelineConfiguration::getId)
-            .collect(Collectors.toSet());
-
+        final Set<String> checkReferencedPipelines = pipelinesWithGeoIpProcessor(clusterState, false);
         if (checkReferencedPipelines.isEmpty()) {
             return false;
         }
@@ -258,22 +256,24 @@ static boolean hasAtLeastOneGeoipProcessor(ClusterState clusterState) {
     }
 
     /**
-     * Retrieve list of pipelines that have at least one geoip processor.
+     * Retrieve the set of pipeline ids that have at least one geoip processor.
      * @param clusterState Cluster state.
      * @param downloadDatabaseOnPipelineCreation Filter the list to include only pipeline with the download_database_on_pipeline_creation
      *                                           matching the param.
-     * @return A list of {@link PipelineConfiguration} matching criteria.
+     * @return A set of pipeline ids matching criteria.
      */
     @SuppressWarnings("unchecked")
-    private static List<PipelineConfiguration> pipelineConfigurationsWithGeoIpProcessor(
-        ClusterState clusterState,
-        boolean downloadDatabaseOnPipelineCreation
-    ) {
-        List<PipelineConfiguration> pipelineDefinitions = IngestService.getPipelines(clusterState);
-        return pipelineDefinitions.stream().filter(pipelineConfig -> {
-            List<Map<String, Object>> processors = (List<Map<String, Object>>) pipelineConfig.getConfigAsMap().get(Pipeline.PROCESSORS_KEY);
-            return hasAtLeastOneGeoipProcessor(processors, downloadDatabaseOnPipelineCreation);
-        }).toList();
+    private static Set<String> pipelinesWithGeoIpProcessor(ClusterState clusterState, boolean downloadDatabaseOnPipelineCreation) {
+        List<PipelineConfiguration> configurations = IngestService.getPipelines(clusterState);
+        Set<String> ids = new HashSet<>();
+        // note: this loop is unrolled rather than streaming-style because it's hot enough to show up in a flamegraph
+        for (PipelineConfiguration configuration : configurations) {
+            List<Map<String, Object>> processors = (List<Map<String, Object>>) configuration.getConfigAsMap().get(Pipeline.PROCESSORS_KEY);
+            if (hasAtLeastOneGeoipProcessor(processors, downloadDatabaseOnPipelineCreation)) {
+                ids.add(configuration.getId());
+            }
+        }
+        return Collections.unmodifiableSet(ids);
     }
 
     /**
@@ -283,7 +283,15 @@ private static List<PipelineConfiguration> pipelineConfigurationsWithGeoIpProces
      * @return true if a geoip processor is found in the processor list.
      */
     private static boolean hasAtLeastOneGeoipProcessor(List<Map<String, Object>> processors, boolean downloadDatabaseOnPipelineCreation) {
-        return processors != null && processors.stream().anyMatch(p -> hasAtLeastOneGeoipProcessor(p, downloadDatabaseOnPipelineCreation));
+        if (processors != null) {
+            // note: this loop is unrolled rather than streaming-style because it's hot enough to show up in a flamegraph
+            for (Map<String, Object> processor : processors) {
+                if (hasAtLeastOneGeoipProcessor(processor, downloadDatabaseOnPipelineCreation)) {
+                    return true;
+                }
+            }
+        }
+        return false;
     }
 
     /**
@@ -317,7 +325,7 @@ private static boolean hasAtLeastOneGeoipProcessor(Map<String, Object> processor
     }
 
     /**
-     * Check if a processor config is has an on_failure clause containing at least a geoip processor.
+     * Check if a processor config has an on_failure clause containing at least a geoip processor.
      * @param processor Processor config.
      * @param downloadDatabaseOnPipelineCreation Should the download_database_on_pipeline_creation of the geoip processor be true or false.
      * @return true if a geoip processor is found in the processor list.
@@ -327,16 +335,17 @@ private static boolean isProcessorWithOnFailureGeoIpProcessor(
         Map<String, Object> processor,
         boolean downloadDatabaseOnPipelineCreation
     ) {
-        return processor != null
-            && processor.values()
-                .stream()
-                .anyMatch(
-                    value -> value instanceof Map
-                        && hasAtLeastOneGeoipProcessor(
-                            ((Map<String, List<Map<String, Object>>>) value).get("on_failure"),
-                            downloadDatabaseOnPipelineCreation
-                        )
-                );
+        // note: this loop is unrolled rather than streaming-style because it's hot enough to show up in a flamegraph
+        for (Object value : processor.values()) {
+            if (value instanceof Map
+                && hasAtLeastOneGeoipProcessor(
+                    ((Map<String, List<Map<String, Object>>>) value).get("on_failure"),
+                    downloadDatabaseOnPipelineCreation
+                )) {
+                return true;
+            }
+        }
+        return false;
     }
 
     /**

modules/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/GeoIpDownloaderTaskExecutor.java

@@ -364,8 +364,7 @@ public BlockLoader blockLoader(BlockLoaderContext blContext) {
             SourceValueFetcher fetcher = SourceValueFetcher.toString(blContext.sourcePaths(name()));
             // MatchOnlyText never has norms, so we have to use the field names field
             BlockSourceReader.LeafIteratorLookup lookup = BlockSourceReader.lookupFromFieldNames(blContext.fieldNames(), name());
-            var sourceMode = blContext.indexSettings().getIndexMappingSourceMode();
-            return new BlockSourceReader.BytesRefsBlockLoader(fetcher, lookup, sourceMode);
+            return new BlockSourceReader.BytesRefsBlockLoader(fetcher, lookup);
         }
 
         @Override

modules/mapper-extras/src/main/java/org/elasticsearch/index/mapper/extras/MatchOnlyTextFieldMapper.java

@@ -319,8 +319,7 @@ public BlockLoader blockLoader(BlockLoaderContext blContext) {
             BlockSourceReader.LeafIteratorLookup lookup = isStored() || isIndexed()
                 ? BlockSourceReader.lookupFromFieldNames(blContext.fieldNames(), name())
                 : BlockSourceReader.lookupMatchingAll();
-            var sourceMode = blContext.indexSettings().getIndexMappingSourceMode();
-            return new BlockSourceReader.DoublesBlockLoader(valueFetcher, lookup, sourceMode);
+            return new BlockSourceReader.DoublesBlockLoader(valueFetcher, lookup);
         }
 
         @Override