Merge branch 'main' into fix-async-stop-result

smalyshev · web-flow · commit a00558e50f34 · 2025-02-07T12:56:55.000-07:00
diff --git a/docs/changelog/121805.yaml b/docs/changelog/121805.yaml
@@ -0,0 +1,5 @@
+pr: 121805
+summary: Support subset of metrics in aggregate metric double
+area: "ES|QL"
+type: enhancement
+issues: []
diff --git a/docs/changelog/122047.yaml b/docs/changelog/122047.yaml
@@ -0,0 +1,5 @@
+pr: 122047
+summary: Fork post-snapshot-delete cleanup off master thread
+area: Snapshot/Restore
+type: bug
+issues: []
diff --git a/libs/entitlement/src/test/java/org/elasticsearch/entitlement/runtime/policy/FileAccessTreeTests.java b/libs/entitlement/src/test/java/org/elasticsearch/entitlement/runtime/policy/FileAccessTreeTests.java
@@ -102,9 +102,6 @@ public void testForwardSlashes() {
         // Forward slashes also work
         assertThat(tree.canRead(path("a/b")), is(true));
         assertThat(tree.canRead(path("m/n")), is(true));
-
-        // In case the native separator is a backslash, don't treat that as an escape
-        assertThat(tree.canRead(path("m\n")), is(false));
     }
 
     FilesEntitlement entitlement(String... values) {
diff --git a/qa/rolling-upgrade/src/javaRestTest/java/org/elasticsearch/upgrades/SourceModeRollingUpgradeIT.java b/qa/rolling-upgrade/src/javaRestTest/java/org/elasticsearch/upgrades/SourceModeRollingUpgradeIT.java
@@ -85,7 +85,13 @@ private void assertDeprecationWarningForTemplate(String templateName) throws IOE
         assertThat(response.containsKey("templates"), equalTo(true));
         Map<?, ?> issuesByTemplate = (Map<?, ?>) response.get("templates");
         assertThat(issuesByTemplate.containsKey(templateName), equalTo(true));
-        var templateIssues = (List<?>) issuesByTemplate.get(templateName);
-        assertThat(((Map<?, ?>) templateIssues.getFirst()).get("message"), equalTo(SourceFieldMapper.DEPRECATION_WARNING));
+        var templateIssue = (Map<?, ?>) ((List<?>) issuesByTemplate.get(templateName)).getFirst();
+        // Bwc compatible logic until backports are complete.
+        if (templateIssue.containsKey("details")) {
+            assertThat(templateIssue.get("message"), equalTo(SourceFieldMapper.DEPRECATION_WARNING_TITLE));
+            assertThat(templateIssue.get("details"), equalTo(SourceFieldMapper.DEPRECATION_WARNING));
+        } else {
+            assertThat(templateIssue.get("message"), equalTo(SourceFieldMapper.DEPRECATION_WARNING));
+        }
     }
 }
diff --git a/server/src/internalClusterTest/java/org/elasticsearch/snapshots/RepositoriesIT.java b/server/src/internalClusterTest/java/org/elasticsearch/snapshots/RepositoriesIT.java
@@ -508,6 +508,12 @@ public void taskSucceeded(ClusterStateTaskListener clusterStateTaskListener, Obj
             .orElseThrow()
             .queue();
 
+        // There is one task in the queue for computing and forking the cleanup work.
+        assertThat(queueLength.getAsInt(), equalTo(1));
+
+        safeAwait(barrier); // unblock the barrier thread and let it process the queue
+        safeAwait(barrier); // wait for the queue to be processed
+
         // There are indexCount (=3*snapshotPoolSize) index-deletion tasks, plus one for cleaning up the root metadata. However, the
         // throttled runner only enqueues one task per SNAPSHOT thread to start with, and then the eager runner adds another one. This shows
         // we are not spamming the threadpool with all the tasks at once, which means that other snapshot activities can run alongside this
diff --git a/server/src/main/java/org/elasticsearch/index/mapper/SourceFieldMapper.java b/server/src/main/java/org/elasticsearch/index/mapper/SourceFieldMapper.java
@@ -58,6 +58,8 @@ public class SourceFieldMapper extends MetadataFieldMapper {
 
     public static final String LOSSY_PARAMETERS_ALLOWED_SETTING_NAME = "index.lossy.source-mapping-parameters";
 
+    public static final String DEPRECATION_WARNING_TITLE = "Configuring source mode in mappings is deprecated.";
+
     public static final String DEPRECATION_WARNING = "Configuring source mode in mappings is deprecated and will be removed "
         + "in future versions. Use [index.mapping.source.mode] index setting instead.";
 
diff --git a/server/src/main/java/org/elasticsearch/repositories/RepositoryData.java b/server/src/main/java/org/elasticsearch/repositories/RepositoryData.java
@@ -29,6 +29,7 @@
 import org.elasticsearch.snapshots.SnapshotInfo;
 import org.elasticsearch.snapshots.SnapshotState;
 import org.elasticsearch.snapshots.SnapshotsService;
+import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
 
@@ -377,6 +378,7 @@ private static boolean isIndexToUpdateAfterRemovingSnapshots(
      * @return map of index to index metadata blob id to delete
      */
     public Map<IndexId, Collection<String>> indexMetaDataToRemoveAfterRemovingSnapshots(Collection<SnapshotId> snapshotIds) {
+        assert ThreadPool.assertCurrentThreadPool(ThreadPool.Names.SNAPSHOT);
         Iterator<IndexId> indicesForSnapshot = indicesToUpdateAfterRemovingSnapshot(snapshotIds);
         final Set<String> allRemainingIdentifiers = indexMetaDataGenerations.lookup.entrySet()
             .stream()
diff --git a/server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java b/server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java
@@ -1134,14 +1134,20 @@ private void runWithUniqueShardMetadataNaming(ActionListener<RepositoryData> rep
                     );
                 })
 
-                .<RepositoryData>andThen((l, newRepositoryData) -> {
-                    l.onResponse(newRepositoryData);
-                    // Once we have updated the repository, run the unreferenced blobs cleanup in parallel to shard-level snapshot deletion
-                    try (var refs = new RefCountingRunnable(onCompletion)) {
-                        cleanupUnlinkedRootAndIndicesBlobs(newRepositoryData, refs.acquireListener());
-                        cleanupUnlinkedShardLevelBlobs(refs.acquireListener());
+                .<RepositoryData>andThen(
+                    // writeIndexGen finishes on master-service thread so must fork here.
+                    snapshotExecutor,
+                    threadPool.getThreadContext(),
+                    (l, newRepositoryData) -> {
+                        l.onResponse(newRepositoryData);
+                        // Once we have updated the repository, run the unreferenced blobs cleanup in parallel to shard-level snapshot
+                        // deletion
+                        try (var refs = new RefCountingRunnable(onCompletion)) {
+                            cleanupUnlinkedRootAndIndicesBlobs(newRepositoryData, refs.acquireListener());
+                            cleanupUnlinkedShardLevelBlobs(refs.acquireListener());
+                        }
                     }
-                })
+                )
 
                 .addListener(repositoryDataUpdateListener);
         }
diff --git a/x-pack/plugin/deprecation/src/main/java/org/elasticsearch/xpack/deprecation/TemplateDeprecationChecker.java b/x-pack/plugin/deprecation/src/main/java/org/elasticsearch/xpack/deprecation/TemplateDeprecationChecker.java
@@ -122,9 +122,9 @@ private DeprecationIssue checkSourceModeInComponentTemplates(ComponentTemplate t
                     if (sourceMap.containsKey("mode")) {
                         return new DeprecationIssue(
                             DeprecationIssue.Level.CRITICAL,
+                            SourceFieldMapper.DEPRECATION_WARNING_TITLE,
+                            "https://ela.st/migrate-source-mode",
                             SourceFieldMapper.DEPRECATION_WARNING,
-                            "https://github.com/elastic/elasticsearch/pull/117172",
-                            null,
                             false,
                             null
                         );
diff --git a/x-pack/plugin/deprecation/src/test/java/org/elasticsearch/xpack/deprecation/TemplateDeprecationCheckerTests.java b/x-pack/plugin/deprecation/src/test/java/org/elasticsearch/xpack/deprecation/TemplateDeprecationCheckerTests.java
@@ -51,9 +51,9 @@ public void testCheckSourceModeInComponentTemplates() throws IOException {
         Map<String, List<DeprecationIssue>> issuesByComponentTemplate = checker.check(clusterState);
         final DeprecationIssue expected = new DeprecationIssue(
             DeprecationIssue.Level.CRITICAL,
+            SourceFieldMapper.DEPRECATION_WARNING_TITLE,
+            "https://ela.st/migrate-source-mode",
             SourceFieldMapper.DEPRECATION_WARNING,
-            "https://github.com/elastic/elasticsearch/pull/117172",
-            null,
             false,
             null
         );
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java
@@ -791,7 +791,12 @@ public enum Cap {
         /**
          * Support for aggregate_metric_double type
          */
-        AGGREGATE_METRIC_DOUBLE(AGGREGATE_METRIC_DOUBLE_FEATURE_FLAG.isEnabled());
+        AGGREGATE_METRIC_DOUBLE(AGGREGATE_METRIC_DOUBLE_FEATURE_FLAG),
+
+        /**
+         * Support for partial subset of metrics in aggregate_metric_double type
+         */
+        AGGREGATE_METRIC_DOUBLE_PARTIAL_SUBMETRICS(AGGREGATE_METRIC_DOUBLE_FEATURE_FLAG);
 
         private final boolean enabled;
 
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/AggregateMapper.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/AggregateMapper.java
@@ -48,7 +48,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -200,9 +199,6 @@ private static Stream<AggDef> aggDefs(Class<?> clazz) {
             if (clazz.isAssignableFrom(Rate.class)) {
                 // rate doesn't support non-grouping aggregations
                 return Stream.of(new AggDef(clazz, type, extra, true));
-            } else if (Objects.equals(type, "AggregateMetricDouble")) {
-                // TODO: support grouping aggregations for aggregate metric double
-                return Stream.of(new AggDef(clazz, type, extra, false));
             } else {
                 return Stream.of(new AggDef(clazz, type, extra, true), new AggDef(clazz, type, extra, false));
             }
diff --git a/x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/mapper/AggregateMetricDoubleFieldMapper.java b/x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/mapper/AggregateMetricDoubleFieldMapper.java
@@ -540,9 +540,6 @@ public AllReader reader(LeafReaderContext context) throws IOException {
                 NumericDocValues sumValues = getNumericDocValues(sumFieldType, context.reader());
                 NumericDocValues valueCountValues = getNumericDocValues(countFieldType, context.reader());
 
-                if (minValues == null || maxValues == null || sumValues == null || valueCountValues == null) {
-                    throw new UnsupportedOperationException("Must have all subfields to use aggregate double metric in ESQL");
-                }
                 return new BlockDocValuesReader() {
 
                     private int docID = -1;
@@ -576,13 +573,13 @@ private void copyDoubleValuesToBuilder(Docs docs, BlockLoader.DoubleBuilder buil
                             if (doc < lastDoc) {
                                 throw new IllegalStateException("docs within same block must be in order");
                             }
-                            if (values.advanceExact(doc)) {
+                            if (values == null || values.advanceExact(doc) == false) {
+                                builder.appendNull();
+                            } else {
                                 double value = NumericUtils.sortableLongToDouble(values.longValue());
                                 lastDoc = doc;
                                 this.docID = doc;
                                 builder.appendDouble(value);
-                            } else {
-                                builder.appendNull();
                             }
                         }
                     }
@@ -595,13 +592,13 @@ private void copyIntValuesToBuilder(Docs docs, BlockLoader.IntBuilder builder, N
                             if (doc < lastDoc) {
                                 throw new IllegalStateException("docs within same block must be in order");
                             }
-                            if (values.advanceExact(doc)) {
+                            if (values == null || values.advanceExact(doc) == false) {
+                                builder.appendNull();
+                            } else {
                                 int value = Math.toIntExact(values.longValue());
                                 lastDoc = doc;
                                 this.docID = doc;
                                 builder.appendInt(value);
-                            } else {
-                                builder.appendNull();
                             }
                         }
                     }
@@ -610,10 +607,10 @@ private void copyIntValuesToBuilder(Docs docs, BlockLoader.IntBuilder builder, N
                     public void read(int docId, StoredFields storedFields, Builder builder) throws IOException {
                         var blockBuilder = (AggregateMetricDoubleBuilder) builder;
                         this.docID = docId;
-                        read(docId, blockBuilder);
+                        readSingleRow(docId, blockBuilder);
                     }
 
-                    private void read(int docId, AggregateMetricDoubleBuilder builder) throws IOException {
+                    private void readSingleRow(int docId, AggregateMetricDoubleBuilder builder) throws IOException {
                         if (minValues.advanceExact(docId)) {
                             builder.min().appendDouble(NumericUtils.sortableLongToDouble(minValues.longValue()));
                         } else {
diff --git a/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/40_tsdb.yml b/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/40_tsdb.yml

Original file line number	Diff line number	Diff line change
`@@ -102,9 +102,6 @@ public void testForwardSlashes() {`
`102`	`102`	`// Forward slashes also work`
`103`	`103`	`assertThat(tree.canRead(path("a/b")), is(true));`
`104`	`104`	`assertThat(tree.canRead(path("m/n")), is(true));`
`105`		`-`
`106`		`- // In case the native separator is a backslash, don't treat that as an escape`
`107`		`- assertThat(tree.canRead(path("m\n")), is(false));`
`108`	`105`	`}`
`109`	`106`
`110`	`107`	`FilesEntitlement entitlement(String... values) {`