[RFC] Introduce Pre-copy Merged Segment into Segment Replication

[guojialiang92]

Is your feature request related to a problem? Please describe

Segment Replication copies the segment held by the primary shard to the replica shard, reducing the build segment overhead of the replica. One of the costs of doing this compared to document replication is the added visibility delay between the primary and the replica. The replica needs to wait for segment replication to complete before searching the doc in the segment.
Currently, the segment replication process includes two types of segments, one generated by merge and the other built by refresh for incremental indexing. Assuming a merged segment is 1GB with a transmission bandwidth of 50MB/s, the segment replication process including this segment will take at least 20s. If there are larger merge segments and more shards occupying the transmission bandwidth, the segment replication process will last longer.

Describe the solution you'd like

Introduction

This RFC introduces the optimization of segment replication, which uses Lucene's IndexWriter.IndexReaderWarmer to pre-copy the merged segment to the replica. It can effectively reduce the delay time before the documents are visible for searching in the replica.

Background

Lucene supports extending IndexWriter.IndexReaderWarmer. After the segment files of merge are generated, IndexWriter.IndexReaderWarmer will be called. The merged segment cannot be searched until the IndexWriter.IndexReaderWarmer is completed. If segment replication is enabled, the merged segment will only participate in the segment replication process after IndexWriter.IndexReaderWarmer is completed.

Proposed Solution

For easy understanding, let's first explain the current segment replication process. segment(_3.si) is generated by segment(_1.si) and segment(_2.si) merge, and segment(_4.si) is generated by refresh. During the segment replication process, they will be replicated together to the replica. If segment(_3.si) is large, the replication process will take a long time, and the docs contained in segment(_4.si) will be invisible to the replica for a long time.

After introducing the Pre-copy Merged Segment, the primary will pre-copy segment(_3.si) to the replica.
Discuss two situations separately:
In the first case, pre-copy is done before segment replication. After refresh, segment(_3.si) and segment(_4.si) are copied to the replica through segment replication. Because the replica already holds the files of segment(_3.si) , these files will be reused during segment replication without network transmission.

The second case is that the pre-copy is done after segment replication. After refresh, the segment(_3.si) is still not visible in the primary, and only the segment(_4.si) is copied to the replica through segment replication.

In the above case, the segment replication process does not include segment(_3.si), reducing the time overhead of segment replication.

Implementation Approaches

Extend IndexReaderWarmer#warm. After the merge generates the files, process as follows:

1. The primary dispatches the merged segment information to all replica nodes
2. The replica pulls the required files from the primary
3. After the replica receives all files or an exception occurs, it returns the response to the primary
4. After the primary receives all replicas' response or times out, it completes the pre-copy merge segment process

Failover

The worst case scenario is that the pre-copy merge segment process encounters an exception or timeout, which will fall back to the current situation. The merged segment is copied to the replica through segment replication.

Related component

Other

Describe alternatives you've considered

No response

Additional context

No response

[shwetathareja]

@guojialiang92 : If I understood your proposal, you are suggesting to copy the merged segment _3.si files first to replicas but dont update the checkpoint to track _3.si yet.

Also, separately transfer of _4.si can be prioritized over _3.si pre-copy so that latest segments are available for search sooner.

Adding @ashking94 to the conversation as he is looking into similar improvements

[guojialiang92]

@guojialiang92 : If I understood your proposal, you are suggesting to copy the merged segment _3.si files first to replicas but dont update the checkpoint to track _3.si yet.

Also, separately transfer of _4.si can be prioritized over _3.si pre-copy so that latest segments are available for search sooner.

Adding @ashking94 to the conversation as he is looking into similar improvements

Yeah, your understanding is correct.
Compared to the current situation, the larger the segment generated by merge, the more obvious the optimization effect of replica visibility.
We will also test the optimization effect of the average delay of segment replication based on the rally test set.

[Bukhtawar]

Thanks, additionally what I was thinking was to maintain separate rate limiter for both merged segments and refreshed segments and prioritise refreshed segments with higher bandwidth and wait async for the merged segments to be available. Also we need to ensure background merges don't pick refreshed segments which haven't been copied over to the replica and maintain that distinction in the merge policy.
Once the refreshed segments haven't been copied and a sort of logical equivalence of segments on the primary/replica established we can update the checkpoints.

I am not sure if the pre-warm logic will slow down searches on primary. Is there a way we could slow down the transmission of merged segments without blocking newly merged segments to be searchable

[ashking94]

Thanks @guojialiang92, we also had faced similar issue with remote store clusters and were starting to work on prioritising the newer segments (due to ingestion) over merged segments for improved data freshness on the replicas for segment replication indexes and remote store clusters.

I had couple of questions regarding your approaches -

In the first case, pre-copy is done before segment replication. After refresh, segment(_3.si) and segment(_4.si) are copied to the replica through segment replication. Because the replica already holds the files of segment(_3.si) , these files will be reused during segment replication without network transmission.

If pre-copy is done before segment replication, wouldn't this still cause the newer docs to be visible on replica after some delay. I believe that _3.si is being copied over while _4.si has been created already. But we would now refresh only after the _3.si is transmitted.

The second case is that the pre-copy is done after segment replication. After refresh, the segment(_3.si) is still not visible in the primary, and only the segment(_4.si) is copied to the replica through segment replication.

Would the reader manager on replica point to _1.si, _2.si and _4.si until the _3.si is being copied over?

[guojialiang92]

Thanks @Bukhtawar

I was thinking was to maintain separate rate limiter for both merged segments and refreshed segments and prioritise refreshed segments with higher bandwidth and wait async for the merged segments to be available.

Strongly agree, configuring different rate limiters for refreshed semgents and merged segments can provide more flexible control.

Also we need to ensure background merges don't pick refreshed segments which haven't been copied over to the replica and maintain that distinction in the merge policy.

Can you give an example? Why do we need to do this?
I think the pre-copy of the merged segment and the copy of the refresh segment are independent processes, and reducing coupling is more ideal.

I am not sure if the pre-warm logic will slow down searches on primary. Is there a way we could slow down the transmission of merged segments without blocking newly merged segments to be searchable

This can be understood as a trade-off between replica shard visibility and primary shard search performance.
In order to avoid that very small segments on primary shard cannot be merged in time, we can trigger the pre-copy logic only when the merged segment exceeds a certain threshold.

[guojialiang92]

Thanks @ashking94

If pre-copy is done before segment replication, wouldn't this still cause the newer docs to be visible on replica after some delay. I believe that _3.si is being copied over while _4.si has been created already. But we would now refresh only after the _3.si is transmitted.

The visibility delay depends only on the transmission time of _4.si. _3.si will be reused，reference org.opensearch.indices.replication.SegmentReplicationTarget#getFiles
If _3.si does not complete the pre-copy before segment replication, it will not participate in segment replication with _4.si (Just like The second case)

Would the reader manager on replica point to _1.si, _2.si and _4.si until the _3.si is being copied over?

Yeah, if pre-copy is not completed for a long time, the reader manager on replica will point to _1.si, _2.si and _4.si.

[Bukhtawar]

I think the pre-copy of the merged segment and the copy of the refresh segment are independent processes, and reducing coupling is more ideal.

Not completely, let me explain there are various reasons why we might see some challenge with the approach you suggest
1/ many systems use a compaction fleet, which are a separate pool of workers decoupled from reader and writers. 2/ one primary can have few slow replicas, any slow replica can potentially slow down the merged segment's freshness. 3/With remote storage as a mode commonly used with segrep, replication checkpoints are just a signal to indicate replicas advance checkpoints independently(no operations on primary block on lagging replicas). So instead of coupling the freshness of the merged segments with the transmission of segments to all copies, can we allow them to happen independently?

Just thinking out aloud, can we not maintain a prioritised list of refreshed segments and mapping of merged segments with individual unmerged segments. Now as soon as each replica catch up on these segments they can swap their checkpoints with merged segments and drop unmerged ones at a slower pace. The refreshed segments can then be prioritised over merged segments

[guojialiang92]

@Bukhtawar
Thank you for your enthusiastic reply!

many systems use a compaction fleet, which are a separate pool of workers decoupled from reader and writers.

Under different usage scenarios, there may be some differences in the working mechanism of pre-copy. There is no remote storage and independent compaction role in the above scheme. I think there will be richer designs for different usage scenarios in the future.

one primary can have few slow replicas, any slow replica can potentially slow down the merged segment's freshness.

We can consider using some strategies for control, such as timeout for pre-copying and segment size for triggering pre-copying.

• Pre-copy timeout will determine the visibility time of the merge segment in the worst case.
• Ensure that small segments are merged in time. For example, pre-copy will only be performed if it exceeds 100MB, merged segments below 100MB can be queried directly.

We are also doing some tests, and the default configuration will depend on the test results.

With remote storage as a mode commonly used with segrep, replication checkpoints are just a signal to indicate replicas advance checkpoints independently(no operations on primary block on lagging replicas).

I think in the case of remote storage, the problem should become simpler. Because the primary shard only needs to upload the merged segment to remote storage, without copying to all replicas. If there are some bottlenecks here, I think they can be considered and resolved separately.

Just thinking out aloud, can we not maintain a prioritised list of refreshed segments and mapping of merged segments with individual unmerged segments. Now as soon as each replica catch up on these segments they can swap their checkpoints with merged segments and drop unmerged ones at a slower pace. The refreshed segments can then be prioritised over merged segments.

Sounds good, achieving this state may require more effort and complexity. The following is what I can think of now:

• The primary shard needs to maintain more state, such as the correspondence between the merge segment and the original segment and refresh segment list. How to manage these states is a new challenge (persistence/failover/cleanup).
• At the same time, the replica's segment_N(segmentInfos) is not consistent with the primary. Different replica shard will point to different segments.

However, in order to solve the problem of high latency in replica visibility, if the query performance is acceptable, is it possible to consider introducing a basic pre-copy process first?