Add segment sorter for data streams

[mayya-sharipova]

It is beneficial to sort segments within a datastream's index
by desc order of their max timestamp field, so
that the most recent (in terms of timestamp) segments
will be first.

This allows to speed up sort query on @timestamp desc field,
which is the most common type of query for datastreams,
as we are mostly concerned with the recent data.
This patch addressed this for writable indices.

Segments' sorter is different from index sorting.
An index sorter by itself is only concerned about the order of docs
within an individual segment (and not how the segments are organized),
while the segment sorter is only used during search and allows
to start docs collection with the "right" segment,
so we can terminate the collection faster.

TODO:

• Should we also do this for frozen or readonly indices?
  Or they always supposed to have 1 segment?

[elasticmachine]

Pinging @elastic/es-search (Team:Search)

[mayya-sharipova]

@elasticmachine run elasticsearch-ci/part-1

[dakrone]

@mayya-sharipova I have a question because I am unfamiliar with this part of the code :) Is this leaf sorter the same as regular index sorting, or is this something that only takes effect on query or merging? Does this affect the indexing throughput the way the index-time sorting can?

[mayya-sharipova]

@dakrone Thank you for taking a look. Answering your questions:

Is this leaf sorter the same as regular index sorting, or is this something that only takes effect on query or merging?

Segments' sorter is different from index sorting, and is only used during querying. An index sorter by itself is only concerned about the order of docs within an individual segment (and not how the segments are organized), while the segment sorter is used during querying to start docs collection with the "right" segment, so we can terminate the collection faster once we collected to the top docs (for data streams, these would be the most recent docs).

Does this affect the indexing throughput the way the index-time sorting can?

No, it doesn't. Segment's sorter is only used during querying.

I've also clarified the PR comment with this information.

[hendrikmuhs]

Does sorting get always applied?

In the case of rollup and transform a composite agg is used and both read data in ascending order (date histogram), in addition both can not benefit from an early exit, but actually disable this explicitly.

That's why I am concerned about this change if it changes the behavior for all cases.

[jimczi]

I left a comment regarding the index metadata change. It should be possible to detect if an index is part of a datastream dynamically.

[jimczi]

I don't think we should change the index metadata for this information. That would only work on data streams created after 7.15 unless I missed something ?
Checking if a concrete index is a data stream can be done dynamically when the IndexShard or IndexService is created ? You can access the index abstraction lookup with clusterService.state().metadata().getIndicesLookup().

[danhermann]

+1 to what @jimczi said here. Individual indices can be detached from and added to data streams in a variety of scenarios including snapshot restores and the relatively new Migrate to Data Stream API. If such a flag were included in index metadata, code would need to be added in all such places to correct the index metadata and we'd prefer to avoid that.

[mayya-sharipova]

@jimczi Thank you for comment. I've started with a dynamic lookup clusterService.state().metadata().getIndicesLookup() like you suggested but it is not allowed to to access a cluster state while modifying cluster state (e.g. during index creation). There is a function ClusterApplierService::assertNotCalledFromClusterStateApplier that checks for it, and returns an error: "the applied cluster state is not yet available". I can investigate if it is possible to do this check somewhere earlier.

@danhermann Thanks for checking, and highlighting Migrate Data Stream API. I would think that this API would be quite rare to use, and most datastream indices would be created in a usual way. But indeed if used, we would need to update index metadata.

I will see if it is possible to do this check dynamically, but as we build more code around datastreams it would be nice to have an easy way to check if a shard is a part of datastream, with APIs like: IndexMetadata::isDataStreamIndex or IndexShard::isDataStreamIndex.

[mayya-sharipova]

@hendrikmuhs Thanks for looking and raising your concerns. Answering your questions:

Does sorting get always applied?

In the case of rollup and transform a composite agg is used and both read data in ascending order (date histogram), in addition both can not benefit from an early exit, but actually disable this explicitly.

Yes, it will be always applied to datastream indices. But this PR is about sorting of segments, in terms of which segment is better to start when doing document collection. It adds a tiny overhead, and for aggregations that collect all documents from all segments, it doesn't really matter from what segment to start, and I don't think there will be any impact on the performance of date histogram or other aggs (but I will make sure to benchmark for this).

I think a bigger concern for your use case could be #71186, which proposes to add sort on @timestamp by default to all searches on datastream indices. Please raise your very valid concerns on that issue.

[hendrikmuhs]

@mayya-sharipova

Thanks for the response. It would be good to benchmark it. If you need help to set this up let me know. It seems tricky to benchmark, I might be wrong, but a lot of benchmarks force-merge to 1 segment before testing, we obviously don't want that behavior in this case.

I have a non-PR'd yet transform track which might be useful, however we don't need transform for this. The aggregation tracks might work as well.

[mayya-sharipova]

@jimczi Thanks for your review so far. In a9d44d9, I've added a data-stream check as you suggested; the check is done in IndicesClusterStateService::createShard, and now it will a property of IndexShard object instead of being a part of stored metadata.

There is an unresolved issue with migrate with data stream API, as there is no closing and reopening of shard during migration, a migration to data stream on already existing migrated indices will not be reflected. We have two options to address it:

1. Decide that it doesn't matter. Migration happens rarely, this only affects speedups in search and on new indices it will be reflected anyway.
2. @dnhatn suggested to apply leaf sorter not only on data stream indices, but on all indices that have @timestamp field. For this, we can check MapperService in a similar way as it is done in TimestampFieldMapperService.

@jimczi WDYT?

[jpountz]

@hendrikmuhs Do I understand you correctly that benchmarking date histograms on @timestamp should be a good enough proxy to make sure that transforms are unaffected by this change? Mayya and I just discussed about doing this before merging this change.

[jimczi]

@dnhatn suggested to apply leaf sorter not only on data stream indices, but on all indices that have @timestamp field. For this, we can check MapperService in a similar way as it is done in TimestampFieldMapperService.

I like the idea. That'll probably require to move the timestamp metadata field in core but that makes sense to me.

[hendrikmuhs]

@hendrikmuhs Do I understand you correctly that benchmarking date histograms on @timestamp should be a good enough proxy to make sure that transforms are unaffected by this change? Mayya and I just discussed about doing this before merging this change.

Yes, that sound like a good plan to me. Thanks for looking into it!

[dimitris-athanasiou]

Another use case that might be affected is scrolling through all docs in ascending order of a date field. This is what ML anomaly detection does when aggregations can't be used (which is many use cases). It might be worth also benchmarking if segment sorting slows this use case down significantly.

[mayya-sharipova]

@elasticmachine run elasticsearch-ci/part-1

[mayya-sharipova]

@elasticmachine run elasticsearch-ci/part-2

[mayya-sharipova]

@elasticmachine run elasticsearch-ci/bwc