Merge branch 'main' into double-question-marks

Author: fang-xing-esql

build-tools-internal/src/main/groovy/elasticsearch.build-scan.gradle

@@ -32,7 +32,9 @@ develocity {
     // Automatically publish scans from Elasticsearch CI
     if (onCI) {
       publishing.onlyIf { true }
-      server = 'https://gradle-enterprise.elastic.co'
+      if(server.isPresent() == false) {
+        server = 'https://gradle-enterprise.elastic.co'
+      }
     } else if( server.isPresent() == false) {
       publishing.onlyIf { false }
     }

docs/changelog/122199.yaml

@@ -0,0 +1,5 @@
+pr: 122199
+summary: Fix issues that prevents using search only snapshots for indices that use index sorting. This is includes Logsdb and time series indices.
+area: Logs
+type: bug
+issues: []

docs/changelog/122496.yaml

@@ -0,0 +1,5 @@
+pr: 122496
+summary: Deduplicate `IngestStats` and `IngestStats.Stats` identity records when deserializing
+area: Ingest Node
+type: bug
+issues: []

muted-tests.yml

@@ -368,12 +368,6 @@ tests:
 - class: org.elasticsearch.xpack.security.authz.IndexAliasesTests
   method: testRemoveIndex
   issue: https://github.com/elastic/elasticsearch/issues/122221
-- class: org.elasticsearch.xpack.esql.action.EsqlActionBreakerIT
-  method: testGroupingMultiValueByOrdinals
-  issue: https://github.com/elastic/elasticsearch/issues/122228
-- class: org.elasticsearch.xpack.esql.action.EsqlNodeFailureIT
-  method: testFailureLoadingFields
-  issue: https://github.com/elastic/elasticsearch/issues/122132
 - class: org.elasticsearch.blocks.SimpleBlocksIT
   method: testConcurrentAddBlock
   issue: https://github.com/elastic/elasticsearch/issues/122324
@@ -385,9 +379,6 @@ tests:
 - class: org.elasticsearch.xpack.esql.action.CrossClusterCancellationIT
   method: testCloseSkipUnavailable
   issue: https://github.com/elastic/elasticsearch/issues/122336
-- class: org.elasticsearch.xpack.esql.action.EsqlActionBreakerIT
-  method: testUnsupportedTypesOrdinalGrouping
-  issue: https://github.com/elastic/elasticsearch/issues/122342
 - class: org.elasticsearch.smoketest.DocsClientYamlTestSuiteIT
   method: test {yaml=reference/alias/line_260}
   issue: https://github.com/elastic/elasticsearch/issues/122343
@@ -398,11 +389,6 @@ 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
-- class: org.elasticsearch.xpack.esql.CsvTests
-  issue: https://github.com/elastic/elasticsearch/issues/122440
 
 # Examples:
 #

server/src/internalClusterTest/java/org/elasticsearch/persistent/PersistentTaskCreationFailureIT.java

@@ -77,7 +77,9 @@ public ClusterState execute(ClusterState currentState) {
                                 .pendingTasks()
                                 .stream()
                                 .filter(
-                                    pendingClusterTask -> pendingClusterTask.getSource().string().equals("finish persistent task (failed)")
+                                    pendingClusterTask -> pendingClusterTask.getSource()
+                                        .string()
+                                        .matches("finish persistent task \\[.*] \\(failed\\)")
                                 )
                                 .count();
                             assertThat(completePersistentTaskPendingTasksCount, lessThanOrEqualTo(1L));

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

@@ -173,11 +173,13 @@ static TransportVersion def(int id) {
     public static final TransportVersion INFERENCE_REQUEST_ADAPTIVE_RATE_LIMITING = def(8_839_0_00);
     public static final TransportVersion ML_INFERENCE_IBM_WATSONX_RERANK_ADDED = def(8_840_0_00);
     public static final TransportVersion COHERE_BIT_EMBEDDING_TYPE_SUPPORT_ADDED_BACKPORT_8_X = def(8_840_0_01);
+    public static final TransportVersion REMOVE_ALL_APPLICABLE_SELECTOR_BACKPORT_8_X = def(8_840_0_02);
     public static final TransportVersion ELASTICSEARCH_9_0 = def(9_000_0_00);
     public static final TransportVersion REMOVE_SNAPSHOT_FAILURES_90 = def(9_000_0_01);
     public static final TransportVersion TRANSPORT_STATS_HANDLING_TIME_REQUIRED_90 = def(9_000_0_02);
     public static final TransportVersion REMOVE_DESIRED_NODE_VERSION_90 = def(9_000_0_03);
     public static final TransportVersion ESQL_DRIVER_TASK_DESCRIPTION_90 = def(9_000_0_04);
+    public static final TransportVersion REMOVE_ALL_APPLICABLE_SELECTOR_9_0 = def(9_000_0_05);
     public static final TransportVersion COHERE_BIT_EMBEDDING_TYPE_SUPPORT_ADDED = def(9_001_0_00);
     public static final TransportVersion REMOVE_SNAPSHOT_FAILURES = def(9_002_0_00);
     public static final TransportVersion TRANSPORT_STATS_HANDLING_TIME_REQUIRED = def(9_003_0_00);

server/src/main/java/org/elasticsearch/action/support/IndexComponentSelector.java

@@ -74,7 +74,9 @@ public static IndexComponentSelector getByKey(String key) {
 
     public static IndexComponentSelector read(StreamInput in) throws IOException {
         byte id = in.readByte();
-        if (in.getTransportVersion().onOrAfter(TransportVersions.REMOVE_ALL_APPLICABLE_SELECTOR)) {
+        if (in.getTransportVersion().onOrAfter(TransportVersions.REMOVE_ALL_APPLICABLE_SELECTOR)
+            || in.getTransportVersion().isPatchFrom(TransportVersions.REMOVE_ALL_APPLICABLE_SELECTOR_9_0)
+            || in.getTransportVersion().isPatchFrom(TransportVersions.REMOVE_ALL_APPLICABLE_SELECTOR_BACKPORT_8_X)) {
             return getById(id);
         } else {
             // Legacy value ::*, converted to ::data

server/src/main/java/org/elasticsearch/common/lucene/Lucene.java

@@ -20,6 +20,8 @@
 import org.apache.lucene.index.ConcurrentMergeScheduler;
 import org.apache.lucene.index.CorruptIndexException;
 import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.FieldInfo;
+import org.apache.lucene.index.FieldInfos;
 import org.apache.lucene.index.FilterCodecReader;
 import org.apache.lucene.index.FilterDirectoryReader;
 import org.apache.lucene.index.FilterLeafReader;
@@ -190,14 +192,26 @@ public static SegmentInfos pruneUnreferencedFiles(String segmentsFileName, Direc
                 throw new IllegalStateException("no commit found in the directory");
             }
         }
+        // Need to figure out what the parent field is that, so that validation in IndexWriter doesn't fail
+        // if no parent field is configured, but FieldInfo says there is a parent field.
+        String parentField = null;
         final IndexCommit cp = getIndexCommit(si, directory);
+        try (var reader = DirectoryReader.open(cp)) {
+            var topLevelFieldInfos = FieldInfos.getMergedFieldInfos(reader);
+            for (FieldInfo fieldInfo : topLevelFieldInfos) {
+                if (fieldInfo.isParentField()) {
+                    parentField = fieldInfo.getName();
+                }
+            }
+        }
         try (
             IndexWriter writer = new IndexWriter(
                 directory,
                 indexWriterConfigWithNoMerging(Lucene.STANDARD_ANALYZER).setSoftDeletesField(Lucene.SOFT_DELETES_FIELD)
                     .setIndexCommit(cp)
                     .setCommitOnClose(false)
                     .setOpenMode(IndexWriterConfig.OpenMode.APPEND)
+                    .setParentField(parentField)
             )
         ) {
             // do nothing and close this will kick off IndexFileDeleter which will remove all pending files

server/src/main/java/org/elasticsearch/index/IndexService.java

@@ -232,7 +232,8 @@ public IndexService(
                 mapperMetrics
             );
             this.indexFieldData = new IndexFieldDataService(indexSettings, indicesFieldDataCache, circuitBreakerService);
-            if (indexSettings.getIndexSortConfig().hasIndexSort()) {
+            boolean sourceOnly = Boolean.parseBoolean(indexSettings.getSettings().get("index.source_only"));
+            if (indexSettings.getIndexSortConfig().hasIndexSort() && sourceOnly == false) {
                 // we delay the actual creation of the sort order for this index because the mapping has not been merged yet.
                 // The sort order is validated right after the merge of the mapping later in the process.
                 this.indexSortSupplier = () -> indexSettings.getIndexSortConfig()

server/src/main/java/org/elasticsearch/ingest/IngestStats.java

@@ -57,22 +57,25 @@ public record IngestStats(Stats totalStats, List<PipelineStat> pipelineStats, Ma
      * Read from a stream.
      */
     public static IngestStats read(StreamInput in) throws IOException {
-        var stats = new Stats(in);
+        var stats = readStats(in);
         var size = in.readVInt();
+        if (stats == Stats.IDENTITY && size == 0) {
+            return IDENTITY;
+        }
         var pipelineStats = new ArrayList<PipelineStat>(size);
         var processorStats = Maps.<String, List<ProcessorStat>>newMapWithExpectedSize(size);
 
         for (var i = 0; i < size; i++) {
             var pipelineId = in.readString();
-            var pipelineStat = new Stats(in);
+            var pipelineStat = readStats(in);
             var byteStat = in.getTransportVersion().onOrAfter(TransportVersions.V_8_15_0) ? new ByteStats(in) : new ByteStats(0, 0);
             pipelineStats.add(new PipelineStat(pipelineId, pipelineStat, byteStat));
             int processorsSize = in.readVInt();
             var processorStatsPerPipeline = new ArrayList<ProcessorStat>(processorsSize);
             for (var j = 0; j < processorsSize; j++) {
                 var processorName = in.readString();
                 var processorType = in.readString();
-                var processorStat = new Stats(in);
+                var processorStat = readStats(in);
                 processorStatsPerPipeline.add(new ProcessorStat(processorName, processorType, processorStat));
             }
             processorStats.put(pipelineId, Collections.unmodifiableList(processorStatsPerPipeline));
@@ -167,20 +170,28 @@ static Map<String, List<ProcessorStat>> merge(Map<String, List<ProcessorStat>> f
         return totalsPerPipelineProcessor;
     }
 
+    /**
+     * Read {@link Stats} from a stream.
+     */
+    private static Stats readStats(StreamInput in) throws IOException {
+        long ingestCount = in.readVLong();
+        long ingestTimeInMillis = in.readVLong();
+        long ingestCurrent = in.readVLong();
+        long ingestFailedCount = in.readVLong();
+        if (ingestCount == 0 && ingestTimeInMillis == 0 && ingestCurrent == 0 && ingestFailedCount == 0) {
+            return Stats.IDENTITY;
+        } else {
+            return new Stats(ingestCount, ingestTimeInMillis, ingestCurrent, ingestFailedCount);
+        }
+    }
+
     public record Stats(long ingestCount, long ingestTimeInMillis, long ingestCurrent, long ingestFailedCount)
         implements
             Writeable,
             ToXContentFragment {
 
         public static final Stats IDENTITY = new Stats(0, 0, 0, 0);
 
-        /**
-         * Read from a stream.
-         */
-        public Stats(StreamInput in) throws IOException {
-            this(in.readVLong(), in.readVLong(), in.readVLong(), in.readVLong());
-        }
-
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             out.writeVLong(ingestCount);