Skip to content
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

[Segment Replication] Support realtime TermVector requests with Segment Replication #9585

Merged
merged 10 commits into from
Sep 8, 2023
Merged

[Segment Replication] Support realtime TermVector requests with Segment Replication #9585

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
4012c16
support realtime TermVector and MultiTermVector requests with segment…
Rishikesh1159 Aug 28, 2023
6f429b6
Merge branch 'main' into fix-vectorTerm-segrep
Rishikesh1159 Aug 29, 2023
2e778ec
Merge branch 'opensearch-project:main' into fix-vectorTerm-segrep
Rishikesh1159 Sep 6, 2023
ae589df
Fix TermVector requests with segrep.
Rishikesh1159 Sep 7, 2023
5d9ccb4
Refacotring.
Rishikesh1159 Sep 7, 2023
b11e4ec
Merge branch 'opensearch-project:main' into fix-vectorTerm-segrep
Rishikesh1159 Sep 7, 2023
89a4588
Merge branch 'opensearch-project:main' into fix-vectorTerm-segrep
Rishikesh1159 Sep 7, 2023
2648fd4
Merge branch 'opensearch-project:main' into fix-vectorTerm-segrep
Rishikesh1159 Sep 7, 2023
b71681f
Address comments on PR.
Rishikesh1159 Sep 8, 2023
4e43c57
Merge branch 'fix-vectorTerm-segrep' of https://github.com/Rishikesh1…
Rishikesh1159 Sep 8, 2023
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
155 changes: 155 additions & 0 deletions ...src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
import org.apache.lucene.document.SortedDocValuesField;
import org.apache.lucene.document.StringField;
import org.apache.lucene.document.TextField;
import org.apache.lucene.index.Fields;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.IndexWriterConfig;
import org.apache.lucene.index.SegmentInfos;
Expand All @@ -38,6 +39,8 @@
import org.opensearch.action.search.SearchResponse;
import org.opensearch.action.search.SearchType;
import org.opensearch.action.support.WriteRequest;
import org.opensearch.action.termvectors.TermVectorsRequestBuilder;
import org.opensearch.action.termvectors.TermVectorsResponse;
import org.opensearch.action.update.UpdateResponse;
import org.opensearch.client.Requests;
import org.opensearch.cluster.ClusterState;
Expand All @@ -57,6 +60,7 @@
import org.opensearch.common.unit.TimeValue;
import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
import org.opensearch.core.index.shard.ShardId;
import org.opensearch.core.xcontent.XContentBuilder;
import org.opensearch.index.IndexModule;
import org.opensearch.index.SegmentReplicationPerGroupStats;
import org.opensearch.index.SegmentReplicationPressureService;
Expand All @@ -81,6 +85,7 @@
import org.opensearch.transport.TransportService;
import org.junit.Before;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
Expand Down Expand Up @@ -1622,4 +1627,154 @@ public void testRealtimeMultiGetRequestsUnsuccessful() {
assertTrue(mgetResponse.getResponses()[1].isFailed());

}

/**
* Tests whether segment replication supports realtime termvector requests and reads and parses source from the translog to serve strong reads.
*/
public void testRealtimeTermVectorRequestsSuccessful() throws IOException {
final String primary = internalCluster().startDataOnlyNode();
XContentBuilder mapping = jsonBuilder().startObject()
.startObject("properties")
.startObject("field")
.field("type", "text")
.field("term_vector", "with_positions_offsets_payloads")
.field("analyzer", "tv_test")
.endObject()
.endObject()
.endObject();
// refresh interval disabled to ensure refresh rate of index (when data is ready for search) doesn't affect realtime termvectors
assertAcked(
prepareCreate(INDEX_NAME).setMapping(mapping)
.addAlias(new Alias("alias"))
.setSettings(
Settings.builder()
.put(indexSettings())
.put("index.analysis.analyzer.tv_test.tokenizer", "standard")
.put("index.refresh_interval", -1)
.putList("index.analysis.analyzer.tv_test.filter", "lowercase")
)
);
final String replica = internalCluster().startDataOnlyNode();
ensureGreen(INDEX_NAME);
final String id = routingKeyForShard(INDEX_NAME, 0);

TermVectorsResponse response = client(replica).prepareTermVectors(indexOrAlias(), "1").get();
assertFalse(response.isExists());

// index doc 1
client().prepareIndex(INDEX_NAME)
.setId(Integer.toString(1))
.setSource(jsonBuilder().startObject().field("field", "the quick brown fox jumps over the lazy dog").endObject())
.execute()
.actionGet();

// non realtime termvectors 1
response = client().prepareTermVectors(indexOrAlias(), Integer.toString(1)).setRealtime(false).get();
assertFalse(response.isExists());

// realtime termvectors 1
TermVectorsRequestBuilder resp = client().prepareTermVectors(indexOrAlias(), Integer.toString(1))
kotwanikunal marked this conversation as resolved.
Show resolved Hide resolved
.setPayloads(true)
.setOffsets(true)
.setPositions(true)
.setRealtime(true)
.setSelectedFields();
response = resp.execute().actionGet();
assertThat(response.getIndex(), equalTo(INDEX_NAME));
assertThat("doc id: " + 1 + " doesn't exists but should", response.isExists(), equalTo(true));
Fields fields = response.getFields();
assertThat(fields.size(), equalTo(1));

// index doc 2 with routing
client().prepareIndex(INDEX_NAME)
.setId(Integer.toString(2))
.setRouting(id)
.setSource(jsonBuilder().startObject().field("field", "the quick brown fox jumps over the lazy dog").endObject())
.execute()
.actionGet();

// realtime termvectors 2 with routing
resp = client().prepareTermVectors(indexOrAlias(), Integer.toString(2))
.setPayloads(true)
.setOffsets(true)
.setPositions(true)
.setRouting(id)
.setSelectedFields();
response = resp.execute().actionGet();
assertThat(response.getIndex(), equalTo(INDEX_NAME));
assertThat("doc id: " + 1 + " doesn't exists but should", response.isExists(), equalTo(true));
fields = response.getFields();
assertThat(fields.size(), equalTo(1));

}

public void testRealtimeTermVectorRequestsUnSuccessful() throws IOException {
final String primary = internalCluster().startDataOnlyNode();
XContentBuilder mapping = jsonBuilder().startObject()
.startObject("properties")
.startObject("field")
.field("type", "text")
.field("term_vector", "with_positions_offsets_payloads")
.field("analyzer", "tv_test")
.endObject()
.endObject()
.endObject();
// refresh interval disabled to ensure refresh rate of index (when data is ready for search) doesn't affect realtime termvectors
assertAcked(
prepareCreate(INDEX_NAME).setMapping(mapping)
.addAlias(new Alias("alias"))
.setSettings(
Settings.builder()
.put(indexSettings())
.put("index.analysis.analyzer.tv_test.tokenizer", "standard")
.put("index.refresh_interval", -1)
.putList("index.analysis.analyzer.tv_test.filter", "lowercase")
.put(indexSettings())
.put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 2)
)
);
final String replica = internalCluster().startDataOnlyNode();
ensureGreen(INDEX_NAME);
final String id = routingKeyForShard(INDEX_NAME, 0);
final String routingOtherShard = routingKeyForShard(INDEX_NAME, 1);

// index doc 1
client().prepareIndex(INDEX_NAME)
.setId(Integer.toString(1))
.setSource(jsonBuilder().startObject().field("field", "the quick brown fox jumps over the lazy dog").endObject())
.setRouting(id)
.execute()
.actionGet();

// non realtime termvectors 1
TermVectorsResponse response = client().prepareTermVectors(indexOrAlias(), Integer.toString(1)).setRealtime(false).get();
assertFalse(response.isExists());

// realtime termvectors (preference = _replica)
TermVectorsRequestBuilder resp = client(replica).prepareTermVectors(indexOrAlias(), Integer.toString(1))
.setPayloads(true)
.setOffsets(true)
.setPositions(true)
.setPreference(Preference.REPLICA.type())
.setRealtime(true)
.setSelectedFields();
response = resp.execute().actionGet();

assertFalse(response.isExists());
assertThat(response.getIndex(), equalTo(INDEX_NAME));

// realtime termvectors (with routing set)
resp = client(replica).prepareTermVectors(indexOrAlias(), Integer.toString(1))
.setPayloads(true)
.setOffsets(true)
.setPositions(true)
.setRouting(routingOtherShard)
.setSelectedFields();
response = resp.execute().actionGet();

assertFalse(response.isExists());
assertThat(response.getIndex(), equalTo(INDEX_NAME));

}

}
14 changes: 12 additions & 2 deletions server/src/main/java/org/opensearch/action/termvectors/TransportTermVectorsAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
import org.opensearch.cluster.ClusterState;
import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
import org.opensearch.cluster.routing.GroupShardsIterator;
import org.opensearch.cluster.routing.Preference;
import org.opensearch.cluster.routing.ShardIterator;
import org.opensearch.cluster.service.ClusterService;
import org.opensearch.common.inject.Inject;
Expand Down Expand Up @@ -87,15 +88,24 @@

@Override
protected ShardIterator shards(ClusterState state, InternalRequest request) {

String preference = request.request().preference;
// For a real time request on a seg rep index, use primary shard as the preferred query shard.
if (request.request().realtime()
&& preference == null
&& state.getMetadata().isSegmentReplicationEnabled(request.concreteIndex())) {
preference = Preference.PRIMARY.type();

Check warning on line 97 in server/src/main/java/org/opensearch/action/termvectors/TransportTermVectorsAction.java

View check run for this annotation

Codecov / codecov/patch

server/src/main/java/org/opensearch/action/termvectors/TransportTermVectorsAction.java#L97 

Added line #L97 was not covered by tests
}

if (request.request().doc() != null && request.request().routing() == null) {
// artificial document without routing specified, ignore its "id" and use either random shard or according to preference
GroupShardsIterator<ShardIterator> groupShardsIter = clusterService.operationRouting()
.searchShards(state, new String[] { request.concreteIndex() }, null, request.request().preference());
.searchShards(state, new String[] { request.concreteIndex() }, null, preference);

Check warning on line 103 in server/src/main/java/org/opensearch/action/termvectors/TransportTermVectorsAction.java

View check run for this annotation

Codecov / codecov/patch

server/src/main/java/org/opensearch/action/termvectors/TransportTermVectorsAction.java#L103 

Added line #L103 was not covered by tests
Copy link
Member

Choose a reason for hiding this comment

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

@Rishikesh1159 I don't think the primary override here is necessary. In this case, an "artificial" document has been supplied and the user is asking to compute term vectors as if this was a real document. I don't think the "real time" parameter is applicable here since it doesn't need to look up any index data.

@msfroh Does that sound right to you?

Copy link
Collaborator

@msfroh msfroh Sep 8, 2023 •
edited
Loading

Choose a reason for hiding this comment

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

That does sound right to me.

I think the override condition should be:

        if (request.request().realtime()
            && preference == null
            && state.getMetadata().isSegmentReplicationEnabled(request.concreteIndex())
            && request.request().doc() == null) {

Or you could revert the edit to this line and continue to use request.request().preference() for the artificial doc case.

Copy link
Member

Choose a reason for hiding this comment

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

Interestingly the code seems to unconditionally attempt to fetch the document even if an artificial document is supplied. Not sure if that complicates things here.

Copy link
Collaborator

@msfroh msfroh Sep 8, 2023 •
edited
Loading

Choose a reason for hiding this comment

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

I think that's probably worth fixing -- from the looks of it, that GetResult is ignored, which seems a little wasteful.

That said, I don't think it hurts anything. In particular, if the doc parameter is specified, then it GETs a document with ID based on an autoincrementing (not really random) integer:

OpenSearch/server/src/main/java/org/opensearch/action/termvectors/TermVectorsRequest.java

Lines 315 to 323 in aca2e9d

public TermVectorsRequest doc(BytesReference doc, boolean generateRandomId, MediaType mediaType) {
// assign a random id to this artificial document, for routing
if (generateRandomId) {
this.id(String.valueOf(randomInt.getAndAdd(1)));
}
this.doc = doc;
this.mediaType = mediaType;
return this;
}

(generateRandomId is always true on the REST parsing code path.)

Copy link
Collaborator

Choose a reason for hiding this comment

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

Still, there is no value in routing to a primary shard if an artificial doc is specified.

return groupShardsIter.iterator().next();
}

return clusterService.operationRouting()
.getShards(state, request.concreteIndex(), request.request().id(), request.request().routing(), request.request().preference());
.getShards(state, request.concreteIndex(), request.request().id(), request.request().routing(), preference);
}

@Override
Expand Down
Loading