Skip to content

Commit 0713c28

Browse files
DonalEvansDonalEvans
authored
[ML] Remove old transport version support from inference classes (#136338)
The oldest supported version that can upgrade to 9.x is 8.18.0, so any classes that version their serialization based on versions older than 8.18.0 no longer need to do so. This commit removes such versioning from classes in xpack/core/inference and xpack/inference. - Remove versioning from writeTo() methods and constructors that take a StreamInput when the version is older than 8.18.0 - Remove related versioning from test methods for affected classes - Remove deprecated LegacyTextEmbeddingResults class
1 parent a9aa4d9 commit 0713c28

24 files changed

+94
-851
lines changed

‎x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/inference/action/DeleteInferenceEndpointAction.java‎

Lines changed: 10 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@
77

88
package org.elasticsearch.xpack.core.inference.action;
99

10-
import org.elasticsearch.TransportVersions;
1110
import org.elasticsearch.action.ActionType;
1211
import org.elasticsearch.action.support.master.AcknowledgedRequest;
1312
import org.elasticsearch.action.support.master.AcknowledgedResponse;
@@ -50,13 +49,8 @@ public Request(StreamInput in) throws IOException {
5049
super(in);
5150
this.inferenceEndpointId = in.readString();
5251
this.taskType = TaskType.fromStream(in);
53-
if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_15_0)) {
54-
this.forceDelete = Boolean.TRUE.equals(in.readOptionalBoolean());
55-
this.dryRun = Boolean.TRUE.equals(in.readOptionalBoolean());
56-
} else {
57-
this.forceDelete = false;
58-
this.dryRun = false;
59-
}
52+
this.forceDelete = Boolean.TRUE.equals(in.readOptionalBoolean());
53+
this.dryRun = Boolean.TRUE.equals(in.readOptionalBoolean());
6054
}
6155

6256
public String getInferenceEndpointId() {
@@ -80,10 +74,8 @@ public void writeTo(StreamOutput out) throws IOException {
8074
super.writeTo(out);
8175
out.writeString(inferenceEndpointId);
8276
taskType.writeTo(out);
83-
if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_15_0)) {
84-
out.writeOptionalBoolean(forceDelete);
85-
out.writeOptionalBoolean(dryRun);
86-
}
77+
out.writeOptionalBoolean(forceDelete);
78+
out.writeOptionalBoolean(dryRun);
8779
}
8880

8981
@Override
@@ -121,32 +113,17 @@ public Response(boolean acknowledged, Set<String> pipelineIds, Set<String> seman
121113

122114
public Response(StreamInput in) throws IOException {
123115
super(in);
124-
if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_15_0)) {
125-
pipelineIds = in.readCollectionAsSet(StreamInput::readString);
126-
} else {
127-
pipelineIds = Set.of();
128-
}
129-
130-
if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_16_0)) {
131-
indexes = in.readCollectionAsSet(StreamInput::readString);
132-
dryRunMessage = in.readOptionalString();
133-
} else {
134-
indexes = Set.of();
135-
dryRunMessage = null;
136-
}
137-
116+
pipelineIds = in.readCollectionAsSet(StreamInput::readString);
117+
indexes = in.readCollectionAsSet(StreamInput::readString);
118+
dryRunMessage = in.readOptionalString();
138119
}
139120

140121
@Override
141122
public void writeTo(StreamOutput out) throws IOException {
142123
super.writeTo(out);
143-
if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_15_0)) {
144-
out.writeCollection(pipelineIds, StreamOutput::writeString);
145-
}
146-
if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_16_0)) {
147-
out.writeCollection(indexes, StreamOutput::writeString);
148-
out.writeOptionalString(dryRunMessage);
149-
}
124+
out.writeCollection(pipelineIds, StreamOutput::writeString);
125+
out.writeCollection(indexes, StreamOutput::writeString);
126+
out.writeOptionalString(dryRunMessage);
150127
}
151128

152129
@Override

‎x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/inference/action/GetInferenceModelAction.java‎

Lines changed: 4 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@
77

88
package org.elasticsearch.xpack.core.inference.action;
99

10-
import org.elasticsearch.TransportVersions;
1110
import org.elasticsearch.action.ActionResponse;
1211
import org.elasticsearch.action.ActionType;
1312
import org.elasticsearch.action.support.master.AcknowledgedRequest;
@@ -19,7 +18,6 @@
1918
import org.elasticsearch.xcontent.XContentBuilder;
2019

2120
import java.io.IOException;
22-
import java.util.ArrayList;
2321
import java.util.List;
2422
import java.util.Objects;
2523

@@ -59,11 +57,7 @@ public Request(StreamInput in) throws IOException {
5957
super(in);
6058
this.inferenceEntityId = in.readString();
6159
this.taskType = TaskType.fromStream(in);
62-
if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_16_0)) {
63-
this.persistDefaultConfig = in.readBoolean();
64-
} else {
65-
this.persistDefaultConfig = PERSIST_DEFAULT_CONFIGS;
66-
}
60+
this.persistDefaultConfig = in.readBoolean();
6761
}
6862

6963
public String getInferenceEntityId() {
@@ -83,9 +77,7 @@ public void writeTo(StreamOutput out) throws IOException {
8377
super.writeTo(out);
8478
out.writeString(inferenceEntityId);
8579
taskType.writeTo(out);
86-
if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_16_0)) {
87-
out.writeBoolean(this.persistDefaultConfig);
88-
}
80+
out.writeBoolean(this.persistDefaultConfig);
8981
}
9082

9183
@Override
@@ -113,12 +105,7 @@ public Response(List<ModelConfigurations> endpoints) {
113105
}
114106

115107
public Response(StreamInput in) throws IOException {
116-
if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_12_0)) {
117-
endpoints = in.readCollectionAsList(ModelConfigurations::new);
118-
} else {
119-
endpoints = new ArrayList<>();
120-
endpoints.add(new ModelConfigurations(in));
121-
}
108+
endpoints = in.readCollectionAsList(ModelConfigurations::new);
122109
}
123110

124111
public List<ModelConfigurations> getEndpoints() {
@@ -127,11 +114,7 @@ public List<ModelConfigurations> getEndpoints() {
127114

128115
@Override
129116
public void writeTo(StreamOutput out) throws IOException {
130-
if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_12_0)) {
131-
out.writeCollection(endpoints);
132-
} else {
133-
endpoints.get(0).writeTo(out);
134-
}
117+
out.writeCollection(endpoints);
135118
}
136119

137120
@Override

‎x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/inference/action/InferenceAction.java‎

Lines changed: 9 additions & 114 deletions
Original file line numberDiff line numberDiff line change
@@ -7,9 +7,7 @@
77

88
package org.elasticsearch.xpack.core.inference.action;
99

10-
import org.elasticsearch.ElasticsearchStatusException;
1110
import org.elasticsearch.TransportVersion;
12-
import org.elasticsearch.TransportVersions;
1311
import org.elasticsearch.action.ActionRequestValidationException;
1412
import org.elasticsearch.action.ActionResponse;
1513
import org.elasticsearch.action.ActionType;
@@ -19,23 +17,16 @@
1917
import org.elasticsearch.common.xcontent.ChunkedToXContentHelper;
2018
import org.elasticsearch.common.xcontent.ChunkedToXContentObject;
2119
import org.elasticsearch.core.TimeValue;
22-
import org.elasticsearch.inference.InferenceResults;
2320
import org.elasticsearch.inference.InferenceServiceResults;
2421
import org.elasticsearch.inference.InputType;
2522
import org.elasticsearch.inference.TaskType;
26-
import org.elasticsearch.rest.RestStatus;
2723
import org.elasticsearch.xcontent.ObjectParser;
2824
import org.elasticsearch.xcontent.ParseField;
2925
import org.elasticsearch.xcontent.ToXContent;
3026
import org.elasticsearch.xcontent.XContentParser;
3127
import org.elasticsearch.xpack.core.inference.InferenceContext;
32-
import org.elasticsearch.xpack.core.inference.results.LegacyTextEmbeddingResults;
33-
import org.elasticsearch.xpack.core.inference.results.SparseEmbeddingResults;
34-
import org.elasticsearch.xpack.core.ml.inference.results.TextExpansionResults;
3528

3629
import java.io.IOException;
37-
import java.util.ArrayList;
38-
import java.util.EnumSet;
3930
import java.util.Iterator;
4031
import java.util.List;
4132
import java.util.Map;
@@ -79,12 +70,6 @@ public static Builder builder(String inferenceEntityId, TaskType taskType) {
7970
PARSER.declareString(Builder::setInferenceTimeout, TIMEOUT);
8071
}
8172

82-
private static final EnumSet<InputType> validEnumsBeforeUnspecifiedAdded = EnumSet.of(InputType.INGEST, InputType.SEARCH);
83-
private static final EnumSet<InputType> validEnumsBeforeClassificationClusteringAdded = EnumSet.range(
84-
InputType.INGEST,
85-
InputType.UNSPECIFIED
86-
);
87-
8873
public static Builder parseRequest(String inferenceEntityId, TaskType taskType, InferenceContext context, XContentParser parser)
8974
throws IOException {
9075
Request.Builder builder = PARSER.apply(parser, null);
@@ -164,25 +149,11 @@ public Request(StreamInput in) throws IOException {
164149
super(in);
165150
this.taskType = TaskType.fromStream(in);
166151
this.inferenceEntityId = in.readString();
167-
if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_12_0)) {
168-
this.input = in.readStringCollectionAsList();
169-
} else {
170-
this.input = List.of(in.readString());
171-
}
152+
this.input = in.readStringCollectionAsList();
172153
this.taskSettings = in.readGenericMap();
173-
if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_13_0)) {
174-
this.inputType = in.readEnum(InputType.class);
175-
} else {
176-
this.inputType = InputType.UNSPECIFIED;
177-
}
178-
179-
if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_14_0)) {
180-
this.query = in.readOptionalString();
181-
this.inferenceTimeout = in.readTimeValue();
182-
} else {
183-
this.query = null;
184-
this.inferenceTimeout = DEFAULT_TIMEOUT;
185-
}
154+
this.inputType = in.readEnum(InputType.class);
155+
this.query = in.readOptionalString();
156+
this.inferenceTimeout = in.readTimeValue();
186157

187158
if (in.getTransportVersion().supports(RERANK_COMMON_OPTIONS_ADDED)) {
188159
this.returnDocuments = in.readOptionalBoolean();
@@ -298,41 +269,18 @@ public void writeTo(StreamOutput out) throws IOException {
298269
super.writeTo(out);
299270
taskType.writeTo(out);
300271
out.writeString(inferenceEntityId);
301-
if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_12_0)) {
302-
out.writeStringCollection(input);
303-
} else {
304-
out.writeString(input.get(0));
305-
}
272+
out.writeStringCollection(input);
306273
out.writeGenericMap(taskSettings);
307-
308-
if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_13_0)) {
309-
out.writeEnum(getInputTypeToWrite(inputType, out.getTransportVersion()));
310-
}
311-
312-
if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_14_0)) {
313-
out.writeOptionalString(query);
314-
out.writeTimeValue(inferenceTimeout);
315-
}
274+
out.writeEnum(inputType);
275+
out.writeOptionalString(query);
276+
out.writeTimeValue(inferenceTimeout);
316277

317278
if (out.getTransportVersion().supports(RERANK_COMMON_OPTIONS_ADDED)) {
318279
out.writeOptionalBoolean(returnDocuments);
319280
out.writeOptionalInt(topN);
320281
}
321282
}
322283

323-
// default for easier testing
324-
static InputType getInputTypeToWrite(InputType inputType, TransportVersion version) {
325-
if (version.before(TransportVersions.V_8_13_0)) {
326-
if (validEnumsBeforeUnspecifiedAdded.contains(inputType) == false) {
327-
return InputType.INGEST;
328-
} else if (validEnumsBeforeClassificationClusteringAdded.contains(inputType) == false) {
329-
return InputType.UNSPECIFIED;
330-
}
331-
}
332-
333-
return inputType;
334-
}
335-
336284
@Override
337285
public boolean equals(Object o) {
338286
if (this == o) return true;
@@ -509,65 +457,12 @@ public Response(InferenceServiceResults results, Flow.Publisher<InferenceService
509457
}
510458

511459
public Response(StreamInput in) throws IOException {
512-
if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_12_0)) {
513-
results = in.readNamedWriteable(InferenceServiceResults.class);
514-
} else {
515-
// It should only be InferenceResults aka TextEmbeddingResults from ml plugin for
516-
// hugging face elser and elser
517-
results = transformToServiceResults(List.of(in.readNamedWriteable(InferenceResults.class)));
518-
}
460+
this.results = in.readNamedWriteable(InferenceServiceResults.class);
519461
// streaming isn't supported via Writeable yet
520462
this.isStreaming = false;
521463
this.publisher = null;
522464
}
523465

524-
@SuppressWarnings("deprecation")
525-
public static InferenceServiceResults transformToServiceResults(List<? extends InferenceResults> parsedResults) {
526-
if (parsedResults.isEmpty()) {
527-
throw new ElasticsearchStatusException(
528-
"Failed to transform results to response format, expected a non-empty list, please remove and re-add the service",
529-
RestStatus.INTERNAL_SERVER_ERROR
530-
);
531-
}
532-
533-
if (parsedResults.get(0) instanceof LegacyTextEmbeddingResults openaiResults) {
534-
if (parsedResults.size() > 1) {
535-
throw new ElasticsearchStatusException(
536-
"Failed to transform results to response format, malformed text embedding result,"
537-
+ " please remove and re-add the service",
538-
RestStatus.INTERNAL_SERVER_ERROR
539-
);
540-
}
541-
542-
return openaiResults.transformToTextEmbeddingResults();
543-
} else if (parsedResults.get(0) instanceof TextExpansionResults) {
544-
return transformToSparseEmbeddingResult(parsedResults);
545-
} else {
546-
throw new ElasticsearchStatusException(
547-
"Failed to transform results to response format, unknown embedding type received,"
548-
+ " please remove and re-add the service",
549-
RestStatus.INTERNAL_SERVER_ERROR
550-
);
551-
}
552-
}
553-
554-
private static SparseEmbeddingResults transformToSparseEmbeddingResult(List<? extends InferenceResults> parsedResults) {
555-
List<TextExpansionResults> textExpansionResults = new ArrayList<>(parsedResults.size());
556-
557-
for (InferenceResults result : parsedResults) {
558-
if (result instanceof TextExpansionResults textExpansion) {
559-
textExpansionResults.add(textExpansion);
560-
} else {
561-
throw new ElasticsearchStatusException(
562-
"Failed to transform results to response format, please remove and re-add the service",
563-
RestStatus.INTERNAL_SERVER_ERROR
564-
);
565-
}
566-
}
567-
568-
return SparseEmbeddingResults.of(textExpansionResults);
569-
}
570-
571466
public InferenceServiceResults getResults() {
572467
return results;
573468
}

0 commit comments

Comments
 (0)