Skip to content

Commit 07f6a8e

Browse files
gaobinlongrayshrey
gaobinlong
authored and
rayshrey
committed
Force merge API supports performing only on primary shards (opensearch-project#11269)
* Force merge API supports performing on primary shards only Signed-off-by: Gao Binlong <[email protected]> * Modify change log Signed-off-by: Gao Binlong <[email protected]> * Fix test failure Signed-off-by: Gao Binlong <[email protected]> * Fix typo Signed-off-by: Gao Binlong <[email protected]> * Modify skip version Signed-off-by: Gao Binlong <[email protected]> * Add version check and more tests Signed-off-by: Gao Binlong <[email protected]> * Format code Signed-off-by: Gao Binlong <[email protected]> * Modify supported version and add more test Signed-off-by: Gao Binlong <[email protected]> * Change the supported version to 3.0.0 Signed-off-by: Gao Binlong <[email protected]> * Add test case in SegmentReplicationIT Signed-off-by: Gao Binlong <[email protected]> * Optimize the test code Signed-off-by: Gao Binlong <[email protected]> --------- Signed-off-by: Gao Binlong <[email protected]>
1 parent b4a77e0 commit 07f6a8e

File tree

13 files changed

+260
-8
lines changed

13 files changed

+260
-8
lines changed

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -104,6 +104,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
104104
- Add toString methods to MultiSearchRequest, MultiGetRequest and CreateIndexRequest ([#12163](https://github.com/opensearch-project/OpenSearch/pull/12163))
105105
- Support for returning scores in matched queries ([#11626](https://github.com/opensearch-project/OpenSearch/pull/11626))
106106
- Add shard id property to SearchLookup for use in field types provided by plugins ([#1063](https://github.com/opensearch-project/OpenSearch/pull/1063))
107+
- Force merge API supports performing on primary shards only ([#11269](https://github.com/opensearch-project/OpenSearch/pull/11269))
107108
- [Tiered caching] Make IndicesRequestCache implementation configurable [EXPERIMENTAL] ([#12533](https://github.com/opensearch-project/OpenSearch/pull/12533))
108109
- Add kuromoji_completion analyzer and filter ([#4835](https://github.com/opensearch-project/OpenSearch/issues/4835))
109110
- The org.opensearch.bootstrap.Security should support codebase for JAR files with classifiers ([#12586](https://github.com/opensearch-project/OpenSearch/issues/12586))

‎rest-api-spec/src/main/resources/rest-api-spec/api/indices.forcemerge.json‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,10 @@
6363
"wait_for_completion": {
6464
"type" : "boolean",
6565
"description" : "If false, the request will return a task immediately and the operation will run in background. Defaults to true."
66+
},
67+
"primary_only": {
68+
"type" : "boolean",
69+
"description" : "Specify whether the operation should only perform on primary shards. Defaults to false."
6670
}
6771
}
6872
}

‎rest-api-spec/src/main/resources/rest-api-spec/test/indices.forcemerge/10_basic.yml‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,3 +27,23 @@
2727
index: test
2828
max_num_segments: 10
2929
only_expunge_deletes: true
30+
31+
---
32+
"Test primary_only parameter":
33+
- skip:
34+
version: " - 2.99.99"
35+
reason: "primary_only is available in 3.0+"
36+
37+
- do:
38+
indices.create:
39+
index: test
40+
body:
41+
settings:
42+
index.number_of_shards: 2
43+
index.number_of_replicas: 1
44+
45+
- do:
46+
indices.forcemerge:
47+
index: test
48+
primary_only: true
49+
- match: { _shards.total: 2 }

‎rest-api-spec/src/main/resources/rest-api-spec/test/indices.forcemerge/20_wait_for_completion.yml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
wait_for_completion: true
2626
task_id: $taskId
2727
- match: { task.action: "indices:admin/forcemerge" }
28-
- match: { task.description: "Force-merge indices [test_index], maxSegments[1], onlyExpungeDeletes[false], flush[true]" }
28+
- match: { task.description: "Force-merge indices [test_index], maxSegments[1], onlyExpungeDeletes[false], flush[true], primaryOnly[false]" }
2929

3030
# .tasks index is created when the force-merge operation completes, so we should delete .tasks index finally,
3131
# if not, the .tasks index may introduce unexpected warnings and then cause other test cases to fail.

‎server/src/internalClusterTest/java/org/opensearch/action/admin/indices/forcemerge/ForceMergeIT.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,24 @@ public void testForceMergeUUIDConsistent() throws IOException {
100100
assertThat(primaryForceMergeUUID, is(replicaForceMergeUUID));
101101
}
102102

103+
public void testForceMergeOnlyOnPrimaryShards() throws IOException {
104+
internalCluster().ensureAtLeastNumDataNodes(2);
105+
final String index = "test-index";
106+
createIndex(
107+
index,
108+
Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1).put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1).build()
109+
);
110+
ensureGreen(index);
111+
final ForceMergeResponse forceMergeResponse = client().admin()
112+
.indices()
113+
.prepareForceMerge(index)
114+
.setMaxNumSegments(1)
115+
.setPrimaryOnly(true)
116+
.get();
117+
assertThat(forceMergeResponse.getFailedShards(), is(0));
118+
assertThat(forceMergeResponse.getSuccessfulShards(), is(1));
119+
}
120+
103121
private static String getForceMergeUUID(IndexShard indexShard) throws IOException {
104122
try (GatedCloseable<IndexCommit> wrappedIndexCommit = indexShard.acquireLastIndexCommit(true)) {
105123
return wrappedIndexCommit.get().getUserData().get(Engine.FORCE_MERGE_UUID_KEY);

‎server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationIT.java‎

Lines changed: 19 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import org.opensearch.action.admin.cluster.stats.ClusterStatsResponse;
2626
import org.opensearch.action.admin.indices.alias.Alias;
2727
import org.opensearch.action.admin.indices.flush.FlushRequest;
28+
import org.opensearch.action.admin.indices.forcemerge.ForceMergeResponse;
2829
import org.opensearch.action.admin.indices.stats.IndicesStatsRequest;
2930
import org.opensearch.action.admin.indices.stats.IndicesStatsResponse;
3031
import org.opensearch.action.get.GetResponse;
@@ -400,6 +401,14 @@ public void testMultipleShards() throws Exception {
400401
}
401402

402403
public void testReplicationAfterForceMerge() throws Exception {
404+
performReplicationAfterForceMerge(false, SHARD_COUNT * (1 + REPLICA_COUNT));
405+
}
406+
407+
public void testReplicationAfterForceMergeOnPrimaryShardsOnly() throws Exception {
408+
performReplicationAfterForceMerge(true, SHARD_COUNT);
409+
}
410+
411+
private void performReplicationAfterForceMerge(boolean primaryOnly, int expectedSuccessfulShards) throws Exception {
403412
final String nodeA = internalCluster().startDataOnlyNode();
404413
final String nodeB = internalCluster().startDataOnlyNode();
405414
createIndex(INDEX_NAME);
@@ -430,8 +439,16 @@ public void testReplicationAfterForceMerge() throws Exception {
430439
waitForDocs(expectedHitCount, indexer);
431440
waitForSearchableDocs(expectedHitCount, nodeA, nodeB);
432441

433-
// Force a merge here so that the in memory SegmentInfos does not reference old segments on disk.
434-
client().admin().indices().prepareForceMerge(INDEX_NAME).setMaxNumSegments(1).setFlush(false).get();
442+
// Perform force merge only on the primary shards.
443+
final ForceMergeResponse forceMergeResponse = client().admin()
444+
.indices()
445+
.prepareForceMerge(INDEX_NAME)
446+
.setPrimaryOnly(primaryOnly)
447+
.setMaxNumSegments(1)
448+
.setFlush(false)
449+
.get();
450+
assertThat(forceMergeResponse.getFailedShards(), is(0));
451+
assertThat(forceMergeResponse.getSuccessfulShards(), is(expectedSuccessfulShards));
435452
refresh(INDEX_NAME);
436453
verifyStoreContent();
437454
}

‎server/src/main/java/org/opensearch/action/admin/indices/forcemerge/ForceMergeRequest.java‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,11 +69,13 @@ public static final class Defaults {
6969
public static final int MAX_NUM_SEGMENTS = -1;
7070
public static final boolean ONLY_EXPUNGE_DELETES = false;
7171
public static final boolean FLUSH = true;
72+
public static final boolean PRIMARY_ONLY = false;
7273
}
7374

7475
private int maxNumSegments = Defaults.MAX_NUM_SEGMENTS;
7576
private boolean onlyExpungeDeletes = Defaults.ONLY_EXPUNGE_DELETES;
7677
private boolean flush = Defaults.FLUSH;
78+
private boolean primaryOnly = Defaults.PRIMARY_ONLY;
7779

7880
private static final Version FORCE_MERGE_UUID_VERSION = Version.V_3_0_0;
7981

@@ -100,6 +102,9 @@ public ForceMergeRequest(StreamInput in) throws IOException {
100102
maxNumSegments = in.readInt();
101103
onlyExpungeDeletes = in.readBoolean();
102104
flush = in.readBoolean();
105+
if (in.getVersion().onOrAfter(Version.V_3_0_0)) {
106+
primaryOnly = in.readBoolean();
107+
}
103108
if (in.getVersion().onOrAfter(FORCE_MERGE_UUID_VERSION)) {
104109
forceMergeUUID = in.readString();
105110
} else if ((forceMergeUUID = in.readOptionalString()) == null) {
@@ -166,6 +171,21 @@ public ForceMergeRequest flush(boolean flush) {
166171
return this;
167172
}
168173

174+
/**
175+
* Should force merge only performed on primary shards. Defaults to {@code false}.
176+
*/
177+
public boolean primaryOnly() {
178+
return primaryOnly;
179+
}
180+
181+
/**
182+
* Should force merge only performed on primary shards. Defaults to {@code false}.
183+
*/
184+
public ForceMergeRequest primaryOnly(boolean primaryOnly) {
185+
this.primaryOnly = primaryOnly;
186+
return this;
187+
}
188+
169189
/**
170190
* Should this task store its result after it has finished?
171191
*/
@@ -188,6 +208,8 @@ public String getDescription() {
188208
+ onlyExpungeDeletes
189209
+ "], flush["
190210
+ flush
211+
+ "], primaryOnly["
212+
+ primaryOnly
191213
+ "]";
192214
}
193215

@@ -197,6 +219,9 @@ public void writeTo(StreamOutput out) throws IOException {
197219
out.writeInt(maxNumSegments);
198220
out.writeBoolean(onlyExpungeDeletes);
199221
out.writeBoolean(flush);
222+
if (out.getVersion().onOrAfter(Version.V_3_0_0)) {
223+
out.writeBoolean(primaryOnly);
224+
}
200225
if (out.getVersion().onOrAfter(FORCE_MERGE_UUID_VERSION)) {
201226
out.writeString(forceMergeUUID);
202227
} else {
@@ -213,6 +238,8 @@ public String toString() {
213238
+ onlyExpungeDeletes
214239
+ ", flush="
215240
+ flush
241+
+ ", primaryOnly="
242+
+ primaryOnly
216243
+ '}';
217244
}
218245
}

‎server/src/main/java/org/opensearch/action/admin/indices/forcemerge/ForceMergeRequestBuilder.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,4 +81,12 @@ public ForceMergeRequestBuilder setFlush(boolean flush) {
8181
request.flush(flush);
8282
return this;
8383
}
84+
85+
/**
86+
* Should force merge only performed on primary shards. Defaults to {@code false}.
87+
*/
88+
public ForceMergeRequestBuilder setPrimaryOnly(boolean primaryOnly) {
89+
request.primaryOnly(primaryOnly);
90+
return this;
91+
}
8492
}

‎server/src/main/java/org/opensearch/action/admin/indices/forcemerge/TransportForceMergeAction.java‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -115,11 +115,16 @@ protected EmptyResult shardOperation(ForceMergeRequest request, ShardRouting sha
115115
}
116116

117117
/**
118-
* The refresh request works against *all* shards.
118+
* The force merge request works against *all* shards by default, but it can work against all primary shards only
119+
* by setting primary_only to true.
119120
*/
120121
@Override
121122
protected ShardsIterator shards(ClusterState clusterState, ForceMergeRequest request, String[] concreteIndices) {
122-
return clusterState.routingTable().allShards(concreteIndices);
123+
if (request.primaryOnly()) {
124+
return clusterState.routingTable().allShardsSatisfyingPredicate(concreteIndices, ShardRouting::primary);
125+
} else {
126+
return clusterState.routingTable().allShards(concreteIndices);
127+
}
123128
}
124129

125130
@Override

‎server/src/main/java/org/opensearch/cluster/routing/RoutingTable.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -307,6 +307,16 @@ public ShardsIterator allShardsSatisfyingPredicate(Predicate<ShardRouting> predi
307307
return allShardsSatisfyingPredicate(indices, predicate, false);
308308
}
309309

310+
/**
311+
* All the shards for the provided indices on the node which match the predicate
312+
* @param indices indices to return all the shards.
313+
* @param predicate condition to match
314+
* @return iterator over shards matching the predicate for the specific indices
315+
*/
316+
public ShardsIterator allShardsSatisfyingPredicate(String[] indices, Predicate<ShardRouting> predicate) {
317+
return allShardsSatisfyingPredicate(indices, predicate, false);
318+
}
319+
310320
private ShardsIterator allShardsSatisfyingPredicate(
311321
String[] indices,
312322
Predicate<ShardRouting> predicate,

0 commit comments

Comments
 (0)