Skip to content

Commit 46f4bce

Browse files
dbwiddisdbwiddis
authored
[RRF] Reject combination techniques the score-ranker-processor cannot use (#1949)
* [RRF] Reject combination techniques the score-ranker-processor cannot use The score-ranker-processor normalizes by rank, so rrf is the only combination technique that means anything to it. RRFProcessorFactory nonetheless accepted any technique registered in ScoreCombinationFactory, including the three means (arithmetic_mean, harmonic_mean, geometric_mean) that belong to the normalization-processor, where they combine already-normalized scores. The resulting pipeline was accepted at creation and then threw NullPointerException on every query: RRFProcessor.recordStats looked the technique up in combTechniqueIncrementers, which maps rrf only, and passed the miss to Optional.of. Since recordStats runs unconditionally before any work, no such pipeline could ever serve a query, so there is no working configuration to preserve. Reject the unsupported technique in the factory, which surfaces as HTTP 400 at pipeline creation instead of HTTP 500 at search time. The message is a superset of the previous one, so an unregistered name such as "my_function" now also reports which technique was rejected and which one is supported. Also switch recordStats to Optional.ofNullable. The factory makes the miss unreachable from a pipeline definition, but the constructor is public, and a stats bookkeeping lookup should never fail a query. Tests: unit coverage for the rejection of each mean and for rrf still being accepted; RRFProcessorTests covers the ofNullable path via direct construction; RRFProcessorIT asserts the 400 and the message against a live cluster rather than assuming the status code. Signed-off-by: Daniel Widdis <widdis@gmail.com> * Add changelog entry Signed-off-by: Daniel Widdis <widdis@gmail.com> --------- Signed-off-by: Daniel Widdis <widdis@gmail.com>
1 parent e975a59 commit 46f4bce

6 files changed

Lines changed: 146 additions & 1 deletion

File tree

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
1313
* [Hybrid Query] Fix NoSuchElementException in hybrid query with sort/search_after when a shard returns no results ([#1939](https://github.com/opensearch-project/neural-search/pull/1939))
1414
* [SemanticHighlighter] Fix SemanticHighlighterExtBuilder.toXContent ([#1906](https://github.com/opensearch-project/neural-search/issues/1906)) (query-insights [#651](https://github.com/opensearch-project/query-insights/issues/651))
1515
* [Sparse ANN] Fold sparse vector tokens into the signed-short range (modulus 32768) so folded tokens are never sign-extended to a negative value when stored in short[] ([#1926](https://github.com/opensearch-project/neural-search/pull/1926))
16+
* [RRF] Reject a combination technique other than rrf when creating a score-ranker-processor, instead of accepting the pipeline and throwing NullPointerException on every query ([#1949](https://github.com/opensearch-project/neural-search/pull/1949))
1617

1718
### Infrastructure
1819

‎src/main/java/org/opensearch/neuralsearch/processor/RRFProcessor.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -156,6 +156,8 @@ <Result extends SearchPhaseResult> Optional<FetchSearchResult> getFetchSearchRes
156156

157157
private void recordStats(ScoreCombinationTechnique combinationTechnique) {
158158
EventStatsManager.increment(EventStatName.RRF_PROCESSOR_EXECUTIONS);
159-
Optional.of(combTechniqueIncrementers.get(combinationTechnique.techniqueName())).ifPresent(Runnable::run);
159+
// ofNullable, not of: a technique with no incrementer must not fail the query. RRFProcessorFactory rejects any
160+
// technique but rrf, so this is no longer reachable from a pipeline definition, but this constructor is public.
161+
Optional.ofNullable(combTechniqueIncrementers.get(combinationTechnique.techniqueName())).ifPresent(Runnable::run);
160162
}
161163
}

‎src/main/java/org/opensearch/neuralsearch/processor/factory/RRFProcessorFactory.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
*/
55
package org.opensearch.neuralsearch.processor.factory;
66

7+
import java.util.Locale;
78
import java.util.Map;
89
import java.util.Objects;
910

@@ -64,6 +65,22 @@ public SearchPhaseResultsProcessor create(
6465
RRFScoreCombinationTechnique.TECHNIQUE_NAME
6566
);
6667

68+
// This processor normalizes by rank, so rrf is the only combination that means anything here. The other
69+
// techniques registered in ScoreCombinationFactory (the means) belong to the normalization-processor, where
70+
// they combine normalized scores. Accepting one here produced a pipeline that threw on every query, so
71+
// reject it at pipeline creation with a 400 instead of failing at search time.
72+
if (RRFScoreCombinationTechnique.TECHNIQUE_NAME.equals(combinationTechnique) == false) {
73+
throw new IllegalArgumentException(
74+
String.format(
75+
Locale.ROOT,
76+
"provided combination technique is not supported by [%s], supported technique is [%s], got [%s]",
77+
RRFProcessor.TYPE,
78+
RRFScoreCombinationTechnique.TECHNIQUE_NAME,
79+
combinationTechnique
80+
)
81+
);
82+
}
83+
6784
String rankConstantParam = RRFNormalizationTechnique.PARAM_NAME_RANK_CONSTANT;
6885
if (combinationClause.containsKey(rankConstantParam)) {
6986
normalizationTechnique = scoreNormalizationFactory.createNormalization(

‎src/test/java/org/opensearch/neuralsearch/processor/RRFProcessorIT.java‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,17 @@
44
*/
55
package org.opensearch.neuralsearch.processor;
66

7+
import com.google.common.collect.ImmutableList;
78
import com.google.common.primitives.Floats;
89
import lombok.SneakyThrows;
910
import lombok.extern.log4j.Log4j2;
11+
import org.apache.hc.core5.http.HttpHeaders;
12+
import org.apache.hc.core5.http.io.entity.EntityUtils;
13+
import org.apache.hc.core5.http.message.BasicHeader;
1014
import org.junit.Before;
15+
import org.opensearch.client.ResponseException;
16+
import org.opensearch.core.rest.RestStatus;
17+
import org.opensearch.common.xcontent.XContentFactory;
1118
import org.opensearch.index.query.MatchQueryBuilder;
1219
import org.opensearch.index.query.QueryBuilder;
1320
import org.opensearch.knn.index.query.KNNQueryBuilder;
@@ -27,6 +34,7 @@
2734
import java.util.stream.Collectors;
2835
import java.util.stream.Stream;
2936

37+
import static org.opensearch.neuralsearch.util.TestUtils.DEFAULT_USER_AGENT;
3038
import static org.opensearch.neuralsearch.util.TestUtils.DELTA_FOR_SCORE_ASSERTION;
3139
import static org.opensearch.neuralsearch.util.TestUtils.TEST_SPACE_TYPE;
3240
import static org.opensearch.neuralsearch.util.TestUtils.createRandomVector;
@@ -57,6 +65,49 @@ public void setUp() throws Exception {
5765
createDefaultRRFSearchPipeline();
5866
}
5967

68+
/**
69+
* The score-ranker-processor normalizes by rank, so rrf is the only combination technique that applies. The mean
70+
* techniques are registered in ScoreCombinationFactory and used to be accepted here, producing a pipeline that
71+
* threw NullPointerException on every query. Creating such a pipeline must fail with 400 instead.
72+
*/
73+
@SneakyThrows
74+
public void testRRFPipelineCreation_whenCombinationTechniqueIsNotRrf_thenBadRequest() {
75+
String body = XContentFactory.jsonBuilder()
76+
.startObject()
77+
.field("description", "score-ranker-processor with a combination technique it does not support")
78+
.startArray("phase_results_processors")
79+
.startObject()
80+
.startObject("score-ranker-processor")
81+
.startObject("combination")
82+
.field("technique", "arithmetic_mean")
83+
.endObject()
84+
.endObject()
85+
.endObject()
86+
.endArray()
87+
.endObject()
88+
.toString();
89+
90+
ResponseException exception = expectThrows(
91+
ResponseException.class,
92+
() -> makeRequest(
93+
client(),
94+
"PUT",
95+
"/_search/pipeline/rrf-unsupported-combination",
96+
null,
97+
toHttpEntity(body),
98+
ImmutableList.of(new BasicHeader(HttpHeaders.USER_AGENT, DEFAULT_USER_AGENT))
99+
)
100+
);
101+
102+
assertEquals(RestStatus.BAD_REQUEST.getStatus(), exception.getResponse().getStatusLine().getStatusCode());
103+
String message = EntityUtils.toString(exception.getResponse().getEntity());
104+
assertTrue(
105+
message,
106+
message.contains("provided combination technique is not supported by [score-ranker-processor], supported technique is [rrf]")
107+
);
108+
assertTrue(message, message.contains("arithmetic_mean"));
109+
}
110+
60111
@SneakyThrows
61112
public void testRRF_whenValidInput_thenSucceed() {
62113
enableStats();

‎src/test/java/org/opensearch/neuralsearch/processor/RRFProcessorTests.java‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import org.opensearch.common.util.concurrent.AtomicArray;
2424
import org.opensearch.core.common.Strings;
2525
import org.opensearch.core.index.shard.ShardId;
26+
import org.opensearch.neuralsearch.processor.combination.ArithmeticMeanScoreCombinationTechnique;
2627
import org.opensearch.neuralsearch.processor.combination.RRFScoreCombinationTechnique;
2728
import org.opensearch.neuralsearch.processor.combination.ScoreCombinationTechnique;
2829
import org.opensearch.neuralsearch.processor.normalization.ScoreNormalizationTechnique;
@@ -138,6 +139,26 @@ public void testProcess_whenValidNonHybridInput_thenSucceed() {
138139
verify(mockNormalizationWorkflow, never()).execute(any(NormalizationProcessorWorkflowExecuteRequest.class));
139140
}
140141

142+
@SneakyThrows
143+
public void testProcess_whenCombinationTechniqueHasNoStatsIncrementer_thenSucceed() {
144+
// RRFProcessorFactory now rejects any combination technique but rrf, so this is unreachable from a pipeline
145+
// definition. It stays reachable by direct construction, and a stats lookup miss must not fail the query.
146+
when(mockCombinationTechnique.techniqueName()).thenReturn(ArithmeticMeanScoreCombinationTechnique.TECHNIQUE_NAME);
147+
QuerySearchResult result = createQuerySearchResult(true);
148+
AtomicArray<SearchPhaseResult> atomicArray = new AtomicArray<>(1);
149+
atomicArray.set(0, result);
150+
151+
when(mockQueryPhaseResultConsumer.getAtomicArray()).thenReturn(atomicArray);
152+
153+
SearchRequest searchRequest = new SearchRequest();
154+
searchRequest.source(new SearchSourceBuilder());
155+
when(mockSearchPhaseContext.getRequest()).thenReturn(searchRequest);
156+
157+
rrfProcessor.process(mockQueryPhaseResultConsumer, mockSearchPhaseContext);
158+
159+
verify(mockNormalizationWorkflow).execute(any(NormalizationProcessorWorkflowExecuteRequest.class));
160+
}
161+
141162
@SneakyThrows
142163
public void testGetTag() {
143164
assertEquals(TAG, rrfProcessor.getTag());

‎src/test/java/org/opensearch/neuralsearch/processor/factory/RRFProcessorFactoryTests.java‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,9 @@
99
import org.opensearch.neuralsearch.processor.NormalizationProcessorWorkflow;
1010
import org.opensearch.neuralsearch.processor.RRFProcessor;
1111
import org.opensearch.neuralsearch.processor.combination.ArithmeticMeanScoreCombinationTechnique;
12+
import org.opensearch.neuralsearch.processor.combination.GeometricMeanScoreCombinationTechnique;
13+
import org.opensearch.neuralsearch.processor.combination.HarmonicMeanScoreCombinationTechnique;
14+
import org.opensearch.neuralsearch.processor.combination.RRFScoreCombinationTechnique;
1215
import org.opensearch.neuralsearch.processor.combination.ScoreCombinationFactory;
1316
import org.opensearch.neuralsearch.processor.combination.ScoreCombiner;
1417
import org.opensearch.neuralsearch.processor.normalization.RRFNormalizationTechnique;
@@ -20,6 +23,7 @@
2023

2124
import java.util.Arrays;
2225
import java.util.HashMap;
26+
import java.util.List;
2327
import java.util.Map;
2428

2529
import static org.mockito.Mockito.mock;
@@ -192,6 +196,55 @@ public void testInvalidCombinationName_whenUnsupportedFunction_thenFail() {
192196
assertTrue(exception.getMessage().contains("provided combination technique is not supported"));
193197
}
194198

199+
@SneakyThrows
200+
public void testInvalidCombinationName_whenTechniqueBelongsToNormalizationProcessor_thenFail() {
201+
// These three are registered in ScoreCombinationFactory, so they used to be accepted here and then threw
202+
// NullPointerException on every query. They combine normalized scores and belong to the normalization-processor;
203+
// this processor normalizes by rank, so rrf is the only combination that applies.
204+
for (String technique : List.of(
205+
ArithmeticMeanScoreCombinationTechnique.TECHNIQUE_NAME,
206+
HarmonicMeanScoreCombinationTechnique.TECHNIQUE_NAME,
207+
GeometricMeanScoreCombinationTechnique.TECHNIQUE_NAME
208+
)) {
209+
RRFProcessorFactory rrfProcessorFactory = new RRFProcessorFactory(
210+
new NormalizationProcessorWorkflow(new ScoreNormalizer(), new ScoreCombiner()),
211+
new ScoreNormalizationFactory(),
212+
new ScoreCombinationFactory()
213+
);
214+
final Map<String, Processor.Factory<SearchPhaseResultsProcessor>> processorFactories = new HashMap<>();
215+
Map<String, Object> config = new HashMap<>();
216+
config.put(COMBINATION_CLAUSE, new HashMap<>(Map.of(TECHNIQUE, technique)));
217+
Processor.PipelineContext pipelineContext = mock(Processor.PipelineContext.class);
218+
219+
IllegalArgumentException exception = expectThrows(
220+
IllegalArgumentException.class,
221+
() -> rrfProcessorFactory.create(processorFactories, "tag", "description", false, config, pipelineContext)
222+
);
223+
assertTrue(
224+
"unexpected message for [" + technique + "]: " + exception.getMessage(),
225+
exception.getMessage()
226+
.contains("provided combination technique is not supported by [score-ranker-processor], supported technique is [rrf]")
227+
);
228+
assertTrue("message should name the rejected technique: " + exception.getMessage(), exception.getMessage().contains(technique));
229+
}
230+
}
231+
232+
@SneakyThrows
233+
public void testCombinationName_whenRrfExplicitlyRequested_thenSuccessful() {
234+
// The counterpart to the rejection above: rrf stated explicitly is still accepted, so the guard rejects only
235+
// the techniques that never worked.
236+
RRFProcessorFactory rrfProcessorFactory = new RRFProcessorFactory(
237+
new NormalizationProcessorWorkflow(new ScoreNormalizer(), new ScoreCombiner()),
238+
new ScoreNormalizationFactory(),
239+
new ScoreCombinationFactory()
240+
);
241+
final Map<String, Processor.Factory<SearchPhaseResultsProcessor>> processorFactories = new HashMap<>();
242+
Map<String, Object> config = new HashMap<>();
243+
config.put(COMBINATION_CLAUSE, new HashMap<>(Map.of(TECHNIQUE, RRFScoreCombinationTechnique.TECHNIQUE_NAME)));
244+
Processor.PipelineContext pipelineContext = mock(Processor.PipelineContext.class);
245+
assertRRFProcessor(rrfProcessorFactory.create(processorFactories, "tag", "description", false, config, pipelineContext));
246+
}
247+
195248
@SneakyThrows
196249
public void testInvalidTechniqueType_whenPassingNormalization_thenSuccessful() {
197250
RRFProcessorFactory rrfProcessorFactory = new RRFProcessorFactory(

0 commit comments

Comments
 (0)