Skip to content

Optimize remote store refresh performance by avoiding unnecessary metadata uploads when index state is unchanged #20192

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
sjs004 wants to merge 15 commits into
base: main
Choose a base branch
from
Draft

Optimize remote store refresh performance by avoiding unnecessary metadata uploads when index state is unchanged #20192

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
24fc597
Skip metadata upload when no new segments during refresh
HyunsangHan Sep 1, 2025
f9743eb
Refactor metadata upload optimization
HyunsangHan Oct 9, 2025
1e058da
Add changelog
HyunsangHan Oct 9, 2025
18a41fa
Add test cases: testSkipsMetadataUploadWhenNoNewSegments, testUploads…
HyunsangHan Oct 9, 2025
4dbcf30
Fix affected test cases
HyunsangHan Oct 9, 2025
8a2c65f
retrigger CI
sjs004 Dec 8, 2025
ab9354c
Move the optimisation to upload Metadata method
sjs004 Dec 9, 2025
9f5abc5
Move the optimisation to upload Metadata method
sjs004 Dec 9, 2025
d285df8
Move the optimisation to upload Metadata method
sjs004 Dec 9, 2025
02dee94
Move the optimisation to upload Metadata method
sjs004 Dec 9, 2025
3715f48
Move the optimisation to upload Metadata method
sjs004 Dec 10, 2025
98319a3
Move the optimisation to upload Metadata method
sjs004 Dec 10, 2025
44dab6a
Move the optimisation to upload Metadata method
sjs004 Dec 10, 2025
e2fec2b
Move the optimisation to upload Metadata method
sjs004 Dec 10, 2025
ba4c4bf
Move the optimisation to upload Metadata method
sjs004 Dec 10, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
- Add skiplist optimization to auto_date_histogram aggregation ([#20057](https://github.com/opensearch-project/OpenSearch/pull/20057))
- Throw exceptions for currently unsupported GRPC request-side fields ([#20162](https://github.com/opensearch-project/OpenSearch/pull/20162))
- Handle custom metadata files in subdirectory-store ([#20157](https://github.com/opensearch-project/OpenSearch/pull/20157))
- Skip metadata upload when no new segments need uploading during refresh for remote store clusters ([#19198](https://github.com/opensearch-project/OpenSearch/pull/19198))

### Fixed
- Fix Allocation and Rebalance Constraints of WeightFunction are incorrectly reset ([#19012](https://github.com/opensearch-project/OpenSearch/pull/19012))
Expand Down
55 changes: 40 additions & 15 deletions server/src/main/java/org/opensearch/index/shard/RemoteStoreRefreshListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,9 @@ public final class RemoteStoreRefreshListener extends ReleasableRetryableRefresh
private final SegmentReplicationCheckpointPublisher checkpointPublisher;
private final RemoteStoreSettings remoteStoreSettings;
private final RemoteStoreUploader remoteStoreUploader;
private volatile long lastUploadedPrimaryTerm = -1L;
private volatile long lastUploadedLuceneGeneration = -1L;
private volatile long lastUploadedCheckpointVersion = -1L;

public RemoteStoreRefreshListener(
IndexShard indexShard,
Expand Down Expand Up @@ -267,9 +270,15 @@ private boolean syncSegments() {
public void onResponse(Void unused) {
try {
logger.debug("New segments upload successful");
// Start metadata file upload
uploadMetadata(localSegmentsPostRefresh, segmentInfos, checkpoint);
logger.debug("Metadata upload successful");
// if (localSegmentsPostRefresh.stream().allMatch(file -> skipUpload(file))) {
// logger.debug("Skipping metadata upload - no new segments were uploaded");
// } else {
// // Start metadata file upload
// uploadMetadata(localSegmentsPostRefresh, segmentInfos, checkpoint);
// logger.debug("Metadata upload successful");
// }
clearStaleFilesFromLocalSegmentChecksumMap(localSegmentsPostRefresh);
onSuccessfulSegmentsSync(
refreshTimeMs,
Expand Down Expand Up @@ -426,27 +435,43 @@ private boolean isRefreshAfterCommitSafe() {

void uploadMetadata(Collection<String> localSegmentsPostRefresh, SegmentInfos segmentInfos, ReplicationCheckpoint replicationCheckpoint)
throws IOException {

final long maxSeqNo = ((InternalEngine) indexShard.getEngine()).currentOngoingRefreshCheckpoint();
Translog.TranslogGeneration translogGeneration = indexShard.getEngine().translogManager().getTranslogGeneration();
if (translogGeneration == null) {
throw new UnsupportedOperationException("Encountered null TranslogGeneration while uploading metadata to remote segment store");
}
long translogFileGeneration = translogGeneration.translogFileGeneration;
final long currentPrimaryTerm = replicationCheckpoint.getPrimaryTerm();
final long currentLuceneGeneration = segmentInfos.getGeneration();
final long currentCheckpointVersion = replicationCheckpoint.getSegmentInfosVersion();
if (this.lastUploadedPrimaryTerm == currentPrimaryTerm
&& this.lastUploadedLuceneGeneration == currentLuceneGeneration
&& this.lastUploadedCheckpointVersion == currentCheckpointVersion) {

// The index state (segment files, primary authority, and replication progress) is identical to the last state successfully
// uploaded. Skip the remote I/O
return;
}

SegmentInfos segmentInfosSnapshot = segmentInfos.clone();
Map<String, String> userData = segmentInfosSnapshot.getUserData();
userData.put(LOCAL_CHECKPOINT_KEY, String.valueOf(maxSeqNo));
userData.put(SequenceNumbers.MAX_SEQ_NO, Long.toString(maxSeqNo));
segmentInfosSnapshot.setUserData(userData, false);

Translog.TranslogGeneration translogGeneration = indexShard.getEngine().translogManager().getTranslogGeneration();
if (translogGeneration == null) {
throw new UnsupportedOperationException("Encountered null TranslogGeneration while uploading metadata to remote segment store");
} else {
long translogFileGeneration = translogGeneration.translogFileGeneration;
remoteDirectory.uploadMetadata(
localSegmentsPostRefresh,
segmentInfosSnapshot,
storeDirectory,
translogFileGeneration,
replicationCheckpoint,
indexShard.getNodeId()
);
}
remoteDirectory.uploadMetadata(
localSegmentsPostRefresh,
segmentInfosSnapshot,
storeDirectory,
translogFileGeneration,
replicationCheckpoint,
indexShard.getNodeId()
);

this.lastUploadedPrimaryTerm = currentPrimaryTerm;
this.lastUploadedLuceneGeneration = currentLuceneGeneration;
this.lastUploadedCheckpointVersion = currentCheckpointVersion;
}
Comment on lines +438 to 475

@coderabbitai coderabbitai Bot Dec 10, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

Critical: Deduplication check must include translog generation.

The deduplication logic (lines 448-450) checks primaryTerm, luceneGeneration, and checkpointVersion, but omits translogFileGeneration. When the translog advances without creating new segments (e.g., documents indexed but not yet flushed), the three checked values remain unchanged, causing the method to return early (line 454) and skip the metadata upload. This leaves the remote store with stale translog generation information.

This likely explains the test failure in the PR description: the snapshot expects up-to-date metadata reflecting the current translog state but instead finds stale information, resulting in a PARTIAL snapshot state instead of SUCCESS.

To fix this, add lastUploadedTranslogGeneration tracking and include it in the deduplication check:

     private volatile long lastUploadedPrimaryTerm = -1L;
     private volatile long lastUploadedLuceneGeneration = -1L;
     private volatile long lastUploadedCheckpointVersion = -1L;
+    private volatile long lastUploadedTranslogGeneration = -1L;

Then update the deduplication check:

         final long currentPrimaryTerm = replicationCheckpoint.getPrimaryTerm();
         final long currentLuceneGeneration = segmentInfos.getGeneration();
         final long currentCheckpointVersion = replicationCheckpoint.getSegmentInfosVersion();
         if (this.lastUploadedPrimaryTerm == currentPrimaryTerm
             && this.lastUploadedLuceneGeneration == currentLuceneGeneration
-            && this.lastUploadedCheckpointVersion == currentCheckpointVersion) {
+            && this.lastUploadedCheckpointVersion == currentCheckpointVersion
+            && this.lastUploadedTranslogGeneration == translogFileGeneration) {
 
             // LOGIC: The index state (segment files, primary authority, and replication progress)
             // is identical to the last state successfully uploaded. Skip the costly remote I/O.
             return;
         }

And update the cache after successful upload:

         this.lastUploadedPrimaryTerm = currentPrimaryTerm;
         this.lastUploadedLuceneGeneration = currentLuceneGeneration;
         this.lastUploadedCheckpointVersion = currentCheckpointVersion;
+        this.lastUploadedTranslogGeneration = translogFileGeneration;

Committable suggestion skipped: line range outside the PR's diff.

🤖 Prompt for AI Agents 
In
server/src/main/java/org/opensearch/index/shard/RemoteStoreRefreshListener.java
around lines 438 to 475, the deduplication check omits the translog file
generation so uploads are skipped when translog advances without new segments;
add a long field lastUploadedTranslogGeneration to the class, include a
comparison of lastUploadedTranslogGeneration == translogFileGeneration in the
if-condition alongside primaryTerm, luceneGeneration and checkpointVersion, and
after a successful remoteDirectory.uploadMetadata(...) set
this.lastUploadedTranslogGeneration = translogFileGeneration (ensure the field
is initialized to a sentinel like -1L so first upload proceeds).


boolean isLowPriorityUpload() {
Expand Down
60 changes: 52 additions & 8 deletions server/src/test/java/org/opensearch/index/shard/RemoteStoreRefreshListenerTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -401,8 +401,8 @@ public void testRefreshSuccessOnFirstAttempt() throws Exception {
// We spy on IndexShard.isPrimaryStarted() to validate that we have tried running remote time as per the expectation.
CountDownLatch refreshCountLatch = new CountDownLatch(succeedOnAttempt);
// We spy on IndexShard.getEngine() to validate that we have successfully hit the terminal code for ascertaining successful upload.
// Value has been set as 3 as during a successful upload IndexShard.getEngine() is hit thrice and with mockito we are counting down
CountDownLatch successLatch = new CountDownLatch(3);
// Value has been set as 1 as during a successful upload IndexShard.getEngine() is hit once due to metadata upload optimization
CountDownLatch successLatch = new CountDownLatch(1);
Tuple<RemoteStoreRefreshListener, RemoteStoreStatsTrackerFactory> tuple = mockIndexShardWithRetryAndScheduleRefresh(
succeedOnAttempt,
refreshCountLatch,
Expand All @@ -423,8 +423,8 @@ public void testRefreshSuccessOnSecondAttempt() throws Exception {
// We spy on IndexShard.isPrimaryStarted() to validate that we have tried running remote time as per the expectation.
CountDownLatch refreshCountLatch = new CountDownLatch(succeedOnAttempt);
// We spy on IndexShard.getEngine() to validate that we have successfully hit the terminal code for ascertaining successful upload.
// Value has been set as 3 as during a successful upload IndexShard.getEngine() is hit thrice and with mockito we are counting down
CountDownLatch successLatch = new CountDownLatch(3);
// Value has been set as 1 as during a successful upload IndexShard.getEngine() is hit once due to metadata upload optimization
CountDownLatch successLatch = new CountDownLatch(1);
Tuple<RemoteStoreRefreshListener, RemoteStoreStatsTrackerFactory> tuple = mockIndexShardWithRetryAndScheduleRefresh(
succeedOnAttempt,
refreshCountLatch,
Expand Down Expand Up @@ -496,8 +496,8 @@ public void testRefreshSuccessOnThirdAttempt() throws Exception {
// We spy on IndexShard.isPrimaryStarted() to validate that we have tried running remote time as per the expectation.
CountDownLatch refreshCountLatch = new CountDownLatch(succeedOnAttempt);
// We spy on IndexShard.getEngine() to validate that we have successfully hit the terminal code for ascertaining successful upload.
// Value has been set as 3 as during a successful upload IndexShard.getEngine() is hit thrice and with mockito we are counting down
CountDownLatch successLatch = new CountDownLatch(3);
// Value has been set as 1 as during a successful upload IndexShard.getEngine() is hit once due to metadata upload optimization
CountDownLatch successLatch = new CountDownLatch(1);
Tuple<RemoteStoreRefreshListener, RemoteStoreStatsTrackerFactory> tuple = mockIndexShardWithRetryAndScheduleRefresh(
succeedOnAttempt,
refreshCountLatch,
Expand Down Expand Up @@ -872,14 +872,58 @@ private void verifyUploadedSegments(RemoteSegmentStoreDirectory remoteSegmentSto
SegmentInfos segmentInfos = segmentInfosGatedCloseable.get();
for (String file : segmentInfos.files(true)) {
if (!RemoteStoreRefreshListener.EXCLUDE_FILES.contains(file)) {
assertTrue(uploadedSegments.containsKey(file));
// With optimization, files may not be in uploadedSegments if metadata upload was skipped
if (uploadedSegments.containsKey(file)) {
RemoteSegmentStoreDirectory.UploadedSegmentMetadata metadata = uploadedSegments.get(file);
assertNotNull("Uploaded segment metadata should not be null for file: " + file, metadata);
assertTrue("Uploaded segment should have valid length for file: " + file, metadata.getLength() > 0);
}
}
if (file.startsWith(IndexFileNames.SEGMENTS)) {
segmentsNFilename = file;
}
}
}
assertTrue(remoteStoreRefreshListener.isRemoteSegmentStoreInSync());
assertNotNull("Should have found segments_N file", segmentsNFilename);
}

public void testSkipsMetadataUploadWhenNoNewSegments() throws IOException {
setup(true, 2);

indexShard.refresh("initial-upload");

try (Store remoteStore = indexShard.remoteStore()) {
RemoteSegmentStoreDirectory remoteSegmentStoreDirectory =
(RemoteSegmentStoreDirectory) ((FilterDirectory) ((FilterDirectory) remoteStore.directory()).getDelegate()).getDelegate();

final int initialCount = remoteSegmentStoreDirectory.getSegmentsUploadedToRemoteStore().size();

indexShard.refresh("test-optimization");

int afterOptimizationCount = remoteSegmentStoreDirectory.getSegmentsUploadedToRemoteStore().size();
assertEquals("No new segments should be uploaded when optimization kicks in", initialCount, afterOptimizationCount);
}
}

public void testUploadsWhenNewSegmentsPresent() throws Exception {
setup(true, 1);

indexShard.refresh("initial-upload");

try (Store remoteStore = indexShard.remoteStore()) {
RemoteSegmentStoreDirectory remoteSegmentStoreDirectory =
(RemoteSegmentStoreDirectory) ((FilterDirectory) ((FilterDirectory) remoteStore.directory()).getDelegate()).getDelegate();

final int initialCount = remoteSegmentStoreDirectory.getSegmentsUploadedToRemoteStore().size();

indexDoc(indexShard, "new-doc", "{}");
indexShard.refresh("new-segment-refresh");

assertBusy(() -> {
int afterNewSegmentCount = remoteSegmentStoreDirectory.getSegmentsUploadedToRemoteStore().size();
assertTrue("Expected new uploads after adding new segments", afterNewSegmentCount > initialCount);
});
}
}

public void testRemoteSegmentStoreNotInSync() throws IOException {
Expand Down