[Intra-SegmentConcurrentSearch] Slicing mechanism

[expani]

Overview

OpenSearch currently distributes the segments into slices in a round robin fashion after sorting them based on their docIds. See MaxTargetSliceSupplier for the implementation. The core idea is

More number of docs in a segment = more work done by the query

The current approach can lead to unequal distribution amongst the slices which is being addressed by #18451 ensuring there is equal distribution as much as possible.

This issue tracks the changes required in slicing mechanism to keep equal distribution while also splitting a segment into multiple partitions.

Meta #18852

Describe the solution you'd like

We are planning to introduce search.concurrent_intra_segment.partition_size discussed more in #18849
This will control the minimum number of docs that should be present in the partition of a segment.

Consider one segment with 31_000 docs and slice count of 3, if search.concurrent_intra_segment.partition_size is equal to 10_000 then the slices will be generated as

[ Slice-1, Slice-2, Slice-3 ]
[ 10_334, 10_333, 10_333 ]

EDIT : Please see the comment below #18851 (comment) for the new proposed approach. The one mentioned below is stale and no longer considered.

To achieve equal distribution with intra segment slicing, the following things need to be done :

• Given all the leaves List<LeafReaferContext> split any leaf which is eligible based on search.concurrent_intra_segment.partition_size

• Generate all List<LeafReaderContextPartition> containing both whole segments and segments which were split into partitions.

• Sort the above list based on the maximum docs and distribute equally amongst the leaf slices using PriorityQueue.

• There could be a case that a slice has 2 partitions of the same leaf/segment in which case IndexSearcher would enforce distinct partitions and throw an exception We would need to merge multiple partitions in a Slice to avoid the exception.

• Also, merging 2 partitions with a slice would be easy but we would need to a pass again to ensure the overall doc count is distributed properly. It could be that for equal distribution 2 partitions that are not continuous fall within same slice.

Related component

Search

[jainankitk]

Thanks @expani for creating this on slicing mechanism for intra segment concurrent search. I am slightly confused about the search.concurrent_intra_segment.partition_size setting. Are we planning to partition segment based on this setting irrespective of the number of slices we are aiming for?

Given some of the redundant work that intra segment search does when using the indexed data, one of the key objectives in my opinion should be to minimize the number of overall partitions across different segments. Keeping that in mind, there are primarily 2 aspects here - the number of documents per segment partition, and total number of documents per slice. We should be using the total number of documents per slice as the maximum number of documents per partition. That should ensure only some of the biggest segments get partitioned and we don't really face the below issue:

There could be a case that a slice has 2 partitions of the same leaf/segment in which case IndexSearcher would enforce distinct partitions and throw an exception We would need to merge multiple partitions in a Slice to avoid the exception.

[expani]

Thanks for taking a look @jainankitk

Are we planning to partition segment based on this setting irrespective of the number of slices we are aiming for?

It will also be capped by the number of slices.

This part in the PR handles the same https://github.com/opensearch-project/OpenSearch/pull/18879/files

        // Max number of splits/partitions for a segment should not exceed the available slices.
        if (numPartitions > targetMaxSlice) {
            numPartitions = targetMaxSlice;
            partitionSize = segmentMaxDoc / numPartitions;
        }

my opinion should be to minimize the number of overall partitions across different segments

Agreed. The approach first achieves equal distribution then moves towards this by merging partitions within the same slice.

We should be using the total number of documents per slice as the maximum number of documents per partition. That should ensure only some of the biggest segments get partitioned and we don't really face the below issue:

I didn't quite get this, would need help elaborating with an example.

The motivation behind allowing user to control the partitionSize ( note: this will be later be added as an execution hint as well ) was to support 2 different types of cases :

• Simple query over huge segments
• Complex query ( Deeply nested boolean query ) over small segments

[atris]

Thank you for this proposal. Some comments:

1. How will you handle the "non-contiguous partition" merging mentioned? Can you provide concrete examples?
2. What's the expected overhead of the additional sorting and merging steps compared to current segment-level distribution?
3. How does partition_size interact with existing maxSliceCount settings?
4. What happens when partitioning creates more overhead than benefit (e.g., very small segments)?

Some suggestions:

1. After merging partitions within slices, could we use incremental heap rebalancing instead of full redistribution and only adjust affected slices? I am assuming the second pass mentioned in your proposal is a redistribution.

2. Could we consider calculating target documents per slice first, then partition segments to fit slice budgets?

3. Could we extend the existing LPT logic in MaxTargetSliceSupplier to handle LeafReaderContextPartition objects?

How does this interact with the existing SliceInputConfig class? The current PR #18879 already modifies MaxTargetSliceSupplier - ensure coordination between these efforts.

[expani]

Background

The core tenet used while discussing solutions for optimal distribution of segments or it's partition among leaf slices is

The more number of docs a segment has the more work a query has to do on that segment

This will always not be true. Consider, a point range query that can a skip a big segment entirely because the range doesn't fall within it. But, at IndexSearcher this information is unavailable or can be expensive to check per query before executing it.

We moved from using Round robin distribution of segments towards using Longest Processing Time ( LPT) algorithm to distribute segments amongst the leaf slices in #18451

LPT is a good enough heuristic for distribution and is always better than round robin. Consider, 5 segments with following doc counts [8,7,6,5,4] and it needs to divided into 2 slices.

LPT will achieve the distribution as Slice-1[8,5,4] and Slice-2[7,6] whereas a better distribution can be done as Slice-1[8,7] and Slice-2[6,5,4]

In Theoretical Computer Science, this problem can be stated as follows

Given a list of items, we need to pack them into a fixed number of bins with each bin having an unlimited size. An item can be fragmented but not below a given minimum size.

Item : Segment ( LeafReaderContext )
Bin : Leaf Slice
Fragment : Partition of a Segment ( LeafReaderContextPartition )

Achieving the most equal distribution (Slice-1[8,7] and Slice-2[6,5,4]) requires algorithms with non-polynomial time complexities ( like Branch and Bound or Integer Programming ) and can end up taking more time than the actual Shard level request.

So, the requirement with Intra Segment Search is :

How can we improve the distribution of segment into slices with LPT given we have the ability to partition a segment ? 

How small can a SegmentPartition (a.k.a. LeafReaderContextPartition) be ?

Every partition of a segment has some fixed cost associated to it which is the initialization of the leaf collector and bulk scorer. If this fixed cost is less than the cost for matching/scoring the documents in that partition, then it's fair to partition the segment for more parallelism.

It is essential we don't create more partitions than the number of slices as having more than one partition of the same segment on the same slice will cause Lucene to throw an error.

The cost of initialization of BulkScorer is dependent on the type of query. Similarly, the cost for initialization of a LeafCollector can be heavily influenced by the underlying aggregation used.

If the search request is a simple term/range query, then SegmentPartition could be a higher value like 50_000 or 100_000 whereas if the search request contains a very complex deeply nested boolean query or contains some heavy-weight script operations a lower value like 1000 would also make sense.

We can always try to give sensible defaults to End User based on workloads like Big5, NYC Taxis, Http Logs; etc. It is essential that the End user can control it via various knobs as well for their workload :

• Cluster Setting
• Index Setting
• Execution Hint

Solution

The proposed solution attempts at using existing LPT based distribution but leverages the ability to partition segments to make the distribution more equal.

Consider a Shard with the 5 segments and their doc counts :

[ 15_000, 4_000, 19_000, 1_000_000, 500_000 ] 

And the minimum partition size is 10_000 ( either from explicit config or default doesn't matter )

The target slice count is 2.

We first split all segments to their minimum partition size. This is to make sure LPT can achieve optimal distribution.

[ 15_000, 4_000, 19_000, 10_000 ..... 100 times, 10_000 ..... 50 times ] 

This will blow up the in-memory size from 5 integers to 153 integers. So, to make the representation concise and ensure sorting takes less time, we can represent each segment as a composite object.

class SegmentPartition {
  boolean isPartitioned;
  int totalDocs;
  int partitionSize; --> This is duplicated for all partitioned segments just kept for representation. 
  int partitionsLeft;
}

[ false, 15_000, -1, 0 ], [ false, 4_000, -1, 0 ], [ false, 19_000, -1, 0 ], [true, 1_000_000, 10_000, 100 ], [true, 500_000, 10_000, 50 ]

During sorting, a custom comparator can consider the totalDocs for segments that are not partitioned whereas it will consider the partitionSize for segments that are partitioned. The sorted segment partitions will look like

[ false, 19_000, -1, 0 ], [ false, 15_000, -1, 0 ], [true, 1_000_000, 10_000, 100 ], [true, 500_000, 10_000, 50 ], [ false, 4_000, -1, 0 ]

With this we have successfully removed the overhead of materializing individual partitions and any increase in complexity for sorting them.

The next challenge is if we assign the partitions individually into PriorityQueue of LeafSlices ( ex; 100 times for segment with 1 million docs ) using LPT then we still end up increasing the overall query time.

To overcome this, we can use simple division to ensure equal distribution.

Slices after first 2 non-partitioned segments are assigned

Slice-1(19_000) Slice-2(15_000)

The slices here won't be Lucene's LeafSlice class rather a custom class like

class LeafSliceBuilder { 
   int totalDocs;
   // This will used later to materialize the actual LeafReaderContextPartition.
   Map<Integer, Integer> segmentOrdToDocCountMap;
}

Now the segment with 1 million docs will try to first increase any slice upto or just less than the largest slice by assigning individual partitions. In this case, it's not possible as assigning 10_000 to Slice-2 will make it more than Slice-1. Generally, this should be equal to the (number of slices - 1) instead of partitions.

Then, we divide the docs present in Segment with 1 million docs equally amongst all slices

Slice-1([19_000 -> SegOrd2][500_000 -> SegOrd3]), 
Slice-2([15_000 -> SegOrd0][500_000 -> SegOrd3]) 

Slice-1(519_000), Slice-2(515_000) 

If there are still remaining docs, it can be assigned from lowest filled slice which is again equal to number of slices - <something>

We keep repeating the same process for every segment that was partitioned. For non-partitioned one's, assignment to a leaf slice remains the same as existing implementation. We finally end up with

Slice-1([19_000 -> SegOrd2][500_000 -> SegOrd3][250_000 -> SegOrd4]), 
Slice-2([15_000 -> SegOrd0][500_000 -> SegOrd3][250_000 -> SegOrd4][4_000 -> SegOrd1]) 

Slice-1(940_000), Slice-2(940_000) 

Once all segments are assigned, we create a map of every segment ord to their last assigned docId.

Call every LeafSliceBuilder with this map which actually materializes the LeafSlice which Lucene expects to have. It also updates the map so that the next slice can have correct docId. We avoided any remerging of partitions with this.

The implementation will be similar to previously closed PR See LeafSliceBuilder.build

We have now made LPT efficient by giving it smallest possible chunks of a segment without the associated memory overhead of creating them or increase in time complexity of sorting or assigning them into leaf slices.

@atris @jainankitk Let me know your thoughts on the same.

[jainankitk]

@expani - Thanks for the detailed explanation of the proposed algorithm.

We have now made LPT efficient by giving it smallest possible chunks of a segment without the associated memory overhead of creating them or increase in time complexity of sorting or assigning them into leaf slices.

While I am not really concerned about the time complexity, but rather the complexity of approach we are introducing. Also, I don't like that we might partition segments that is not necessarily required in this approach. I am wondering if there is better and simpler approach. Starting with the example proposed above:

Consider a Shard with the 5 segments and their doc counts :

[ 15_000, 4_000, 19_000, 1_000_000, 500_000 ] 

The target slice count is 2. The total number of documents is 1_538_000, hence max_documents_per_segment_partition = 1_538_000 / 2 ~= 779_000. So, we only need to partition segment 4 into ceiling(1_000_000/779_000) = 2 partitions, as rest of the segments are below this threshold.

Final list of LPT processing becomes:

[ 15_000, 4_000, 19_000, 500_000 (seg4), 500_000 (seg4), 500_000 ]

which when sorted becomes:

[ 500_000 (seg4), 500_000 (seg4), 500_000, 15_000, 4_000, 19_000 ]

and final slices after applying LPT:

Slice1 [500_000 (seg4), 500_000]
Slice2 [500_000 (seg4), 19_000, 15_000, 4_000]

After the final step, we can have optional step to balance between minSlice / maxSlice if the skew is above certain threshold.

Hence, the final result is:

Slice1 [279_000 (seg4), 500_000]
Slice2 [721_000 (seg4), 19_000, 15_000, 4_000]

[expani]

Thanks for mentioning the alternative approach @jainankitk

While I am not really concerned about the time complexity,

I think we need to consider this since we could have used Branch&Bound or some other complex algorithm to solve bin packing most efficiently. But, then it starts taking more time than the actual shard search.

but rather the complexity of approach we are introducing

The code can be written in a way that this only applies for intra segment when enabled. I agree it's a bit more complex to implement but can achieve a better distribution while not increasing distribution time.

Also, I don't like that we might partition segments that is not necessarily required in this approach.

It shouldn't cause an issue if the overall query time reduces since we are ensuring we don't go below a minimum threshold where partitioning doesn't make sense.

Let's consider another case of 5 segments like

[ 800_000, 700_000, 600_000, 500_000, 400_000 ] 

which needs to be divided into 2 slices.

Here the total docs is 3_000_000 and max_docs_per_segment_partition = 1_500_000 so no segments will be split any further and the distribution with LPT will look like

Slice-1 [ 800_000, 500_000, 400_000 ]
Slice-2 [ 700_000, 600_000 ]

leaving a difference of 400_000 docs between the 2 slices. We will end up not reducing the overall query execution time even though it's possible with further splitting.