Expand segment sorter for all timeseries indices (#78639)

mayya-sharipova · web-flow · commit a74573be9ec0 · 2021-10-05T13:21:39.000-04:00
PR #75195 added segment sorter on @timestamp desc for datastream indices. This PR applies segment sorter to all indices that have @timestamp field. The presence of @timestamp field can serve as a strong indication that we are dealing with timeseries indices. The most common type of query for timeseries indices is to get the latest data, that is data sorted by @timestamp desc. This PR sorts segments by @timestamp desc which allows to speed up this kind of queries. Relates to #75195
diff --git a/rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/search/380_sort_segments_on_timestamp.yml b/rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/search/380_sort_segments_on_timestamp.yml
@@ -0,0 +1,139 @@
+---
+"Test that index segments are sorted on timestamp field if @timestamp field is defined in mapping":
+  - skip:
+      version: " - 7.99.99"
+      reason: "sorting segments was added in 7.16"
+      features: allowed_warnings
+
+  - do:
+      indices.create:
+        index: test_index1
+        body:
+          mappings:
+            properties:
+              "@timestamp":
+                type: date
+          settings:
+            number_of_shards: 1
+            number_of_replicas: 0
+
+  # 1st segment
+  - do:
+      index:
+        index: test_index1
+        body: { "foo": "bar1", "@timestamp": "2021-08-01" }
+        refresh: true
+
+  # 2nd segment
+  - do:
+      index:
+        index: test_index1
+        body: { "foo": "bar2", "@timestamp": "2021-08-02" }
+        refresh: true
+
+  # test that segments are sorted by @timestamp DESC
+  - do:
+      search:
+        index: test_index1
+        body:
+          fields: [{ "field":"@timestamp", "format":"yyyy-MM-dd" }]
+  - match: { hits.total.value: 2 }
+  - match: { hits.hits.0.fields.@timestamp: ["2021-08-02"] }
+  - match: { hits.hits.1.fields.@timestamp: ["2021-08-01"] }
+
+---
+"Test that index segments are NOT sorted on timestamp field when @timestamp field is dynamically added":
+  - skip:
+      version: " - 7.99.99"
+      reason: "sorting segments was added in 7.16"
+      features: allowed_warnings
+
+  - do:
+      indices.create:
+        index: test_index2
+        body:
+          settings:
+            number_of_shards: 1
+            number_of_replicas: 0
+
+  # 1st segment
+  - do:
+      index:
+        index: test_index2
+        body: { "foo": "bar1", "@timestamp": "2021-08-01" }
+        refresh: true
+
+  # 2nd segment
+  - do:
+      index:
+        index: test_index2
+        body: { "foo": "bar2", "@timestamp": "2021-08-02" }
+        refresh: true
+
+  # test that segments are NOT sorted by @timestamp DESC as the field was not
+  - do:
+      search:
+        index: test_index2
+        body:
+          fields: [{ "field":"@timestamp", "format":"yyyy-MM-dd" }]
+  - match: { hits.hits.0.fields.@timestamp: ["2021-08-01"] }
+  - match: { hits.hits.1.fields.@timestamp: ["2021-08-02"] }
+
+  # test that after we reopen the index, segments are sorted by @timestamp DESC
+  - do:
+      indices.close:
+        index: test_index2
+  - is_true: acknowledged
+  - do:
+      indices.open:
+        index: test_index2
+  - is_true: acknowledged
+  - do:
+      search:
+        index: test_index2
+        body:
+          fields: [{ "field":"@timestamp", "format":"yyyy-MM-dd" }]
+  - match: { hits.total.value: 2 }
+  - match: { hits.hits.0.fields.@timestamp: ["2021-08-02"] }
+  - match: { hits.hits.1.fields.@timestamp: ["2021-08-01"] }
+
+---
+"Test if segments are missing @timestamp field we don't get errors":
+  - skip:
+      version: " - 7.99.99"
+      reason: "sorting segments was added in 7.16"
+      features: allowed_warnings
+
+  - do:
+      indices.create:
+        index: test_index3
+        body:
+          mappings:
+            properties:
+              "@timestamp":
+                type: date
+          settings:
+            number_of_shards: 1
+            number_of_replicas: 0
+
+  # 1st segment missing @timestamp field
+  - do:
+      index:
+        index: test_index3
+        body: { "foo": "bar1"}
+        refresh: true
+
+  # 2nd segment
+  - do:
+      index:
+        index: test_index3
+        body: { "foo": "bar2", "@timestamp": "2021-08-02" }
+        refresh: true
+
+  - do:
+      search:
+        index: test_index3
+        body:
+          fields: [{ "field":"@timestamp", "format":"yyyy-MM-dd" }]
+  - match: { hits.hits.0.fields.@timestamp: ["2021-08-02"] }
+  - is_false: hits.hits.1.fields.@timestamp
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/DataStream.java b/server/src/main/java/org/elasticsearch/cluster/metadata/DataStream.java
@@ -10,6 +10,7 @@
 import org.apache.lucene.document.LongPoint;
 import org.apache.lucene.index.LeafReader;
 import org.apache.lucene.index.PointValues;
+import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.cluster.AbstractDiffable;
 import org.elasticsearch.cluster.Diff;
 import org.elasticsearch.common.Strings;
@@ -42,23 +43,26 @@ public final class DataStream extends AbstractDiffable<DataStream> implements To
 
     public static final String BACKING_INDEX_PREFIX = ".ds-";
     public static final DateFormatter DATE_FORMATTER = DateFormatter.forPattern("uuuu.MM.dd");
-    // Datastreams' leaf readers should be sorted by desc order of their timestamp field, as it allows search time optimizations
-    public static Comparator<LeafReader> DATASTREAM_LEAF_READERS_SORTER =
+    // Timeseries indices' leaf readers should be sorted by desc order of their timestamp field, as it allows search time optimizations
+    public static Comparator<LeafReader> TIMESERIES_LEAF_READERS_SORTER =
         Comparator.comparingLong(
             (LeafReader r) -> {
                 try {
                     PointValues points = r.getPointValues(DataStream.TimestampField.FIXED_TIMESTAMP_FIELD);
                     if (points != null) {
                         byte[] sortValue = points.getMaxPackedValue();
                         return LongPoint.decodeDimension(sortValue, 0);
-                    } else if (r.numDocs() == 0) {
-                        // points can be null if the segment contains only deleted documents
+                    } else {
+                        // As we apply this segment sorter to any timeseries indices,
+                        // we don't have a guarantee that all docs contain @timestamp field.
+                        // Some segments may have all docs without @timestamp field, in this
+                        // case they will be sorted last.
                         return Long.MIN_VALUE;
                     }
                 } catch (IOException e) {
+                    throw new ElasticsearchException("Can't access [" +
+                    DataStream.TimestampField.FIXED_TIMESTAMP_FIELD + "] field for the index!", e);
                 }
-                throw new IllegalStateException("Can't access [" +
-                    DataStream.TimestampField.FIXED_TIMESTAMP_FIELD + "] field for the data stream!");
             })
         .reversed();
 
diff --git a/server/src/main/java/org/elasticsearch/index/mapper/MappingLookup.java b/server/src/main/java/org/elasticsearch/index/mapper/MappingLookup.java
@@ -9,6 +9,7 @@
 package org.elasticsearch.index.mapper;
 
 import org.apache.lucene.codecs.PostingsFormat;
+import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.analysis.IndexAnalyzers;
 import org.elasticsearch.index.analysis.NamedAnalyzer;
@@ -389,6 +390,19 @@ public boolean isDataStreamTimestampFieldEnabled() {
         return dtfm != null && dtfm.isEnabled();
     }
 
+    /**
+     * Returns if this mapping contains a timestamp field that is of type date, indexed and has doc values.
+     * @return {@code true} if contains a timestamp field of type date that is indexed and has doc values, {@code false} otherwise.
+     */
+    public boolean hasTimestampField() {
+        final MappedFieldType mappedFieldType = fieldTypesLookup().get(DataStream.TimestampField.FIXED_TIMESTAMP_FIELD);
+        if (mappedFieldType instanceof DateFieldMapper.DateFieldType) {
+            return mappedFieldType.isSearchable() && mappedFieldType.hasDocValues();
+        } else {
+            return false;
+        }
+    }
+
     /**
      * Key for the lookup to be used in caches.
      */
diff --git a/server/src/main/java/org/elasticsearch/index/shard/IndexShard.java b/server/src/main/java/org/elasticsearch/index/shard/IndexShard.java
@@ -186,7 +186,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
-import static org.elasticsearch.cluster.metadata.DataStream.DATASTREAM_LEAF_READERS_SORTER;
+import static org.elasticsearch.cluster.metadata.DataStream.TIMESERIES_LEAF_READERS_SORTER;
 import static org.elasticsearch.index.seqno.RetentionLeaseActions.RETAIN_ALL;
 import static org.elasticsearch.index.seqno.SequenceNumbers.UNASSIGNED_SEQ_NO;
 
@@ -408,6 +408,14 @@ public Sort getIndexSort() {
         return indexSortSupplier.get();
     }
 
+    /**
+     * Returns if this shard is a part of datastream
+     * @return {@code true} if this shard is a part of datastream, {@code false} otherwise
+     */
+    public boolean isDataStreamIndex() {
+        return isDataStreamIndex;
+    }
+
     public ShardGetService getService() {
         return this.getService;
     }
@@ -2905,6 +2913,7 @@ private EngineConfig newEngineConfig(LongSupplier globalCheckpointSupplier) {
                 this.warmer.warm(reader);
             }
         };
+        final boolean isTimeseriesIndex = mapperService == null ? false : mapperService.mappingLookup().hasTimestampField();
         return new EngineConfig(
                 shardId,
                 threadPool,
@@ -2928,7 +2937,7 @@ private EngineConfig newEngineConfig(LongSupplier globalCheckpointSupplier) {
                 replicationTracker::getRetentionLeases,
                 this::getOperationPrimaryTerm,
                 snapshotCommitSupplier,
-                isDataStreamIndex ? DATASTREAM_LEAF_READERS_SORTER : null);
+                isTimeseriesIndex ? TIMESERIES_LEAF_READERS_SORTER : null);
     }
 
     /**
diff --git a/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/data_stream/131_sort_segments_migrate_to_data_stream.yml b/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/data_stream/131_sort_segments_migrate_to_data_stream.yml
diff --git a/x-pack/qa/runtime-fields/build.gradle b/x-pack/qa/runtime-fields/build.gradle
@@ -96,6 +96,8 @@ subprojects {
           // The error messages are different
           'search/330_fetch_fields/error includes field name',
           'search/330_fetch_fields/error includes glob pattern',
+          // we need a @timestamp field to be defined in index mapping
+          'search/380_sort_segments_on_timestamp/*',
           /////// NOT SUPPORTED ///////
         ].join(',')
     }

Original file line number	Diff line number	Diff line change
`@@ -96,6 +96,8 @@ subprojects {`
`96`	`96`	`// The error messages are different`
`97`	`97`	`'search/330_fetch_fields/error includes field name',`
`98`	`98`	`'search/330_fetch_fields/error includes glob pattern',`
	`99`	`+ // we need a @timestamp field to be defined in index mapping`
	`100`	`+ 'search/380_sort_segments_on_timestamp/*',`
`99`	`101`	`/////// NOT SUPPORTED ///////`
`100`	`102`	`].join(',')`
`101`	`103`	`}`