Block hybrid query when no search pipeline is configured #1924
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -14,21 +14,31 @@ | |
| import org.opensearch.action.search.SearchType; | ||
| import org.opensearch.action.support.ActionFilter; | ||
| import org.opensearch.action.support.ActionFilterChain; | ||
| import org.opensearch.cluster.metadata.IndexMetadata; | ||
| import org.opensearch.core.action.ActionResponse; | ||
| import org.opensearch.index.IndexSettings; | ||
| import org.opensearch.index.query.QueryBuilder; | ||
| import org.opensearch.neuralsearch.query.HybridQueryBuilder; | ||
| import org.opensearch.neuralsearch.util.HybridQueryUtil; | ||
| import org.opensearch.neuralsearch.util.NeuralSearchClusterUtil; | ||
| import org.opensearch.search.pipeline.SearchPipelineService; | ||
| import org.opensearch.tasks.Task; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| import lombok.extern.log4j.Log4j2; | ||
|
|
||
| /** | ||
| * An ActionFilter that validates hybrid query requests and disables batched reduction for them. | ||
| * | ||
| * This filter intercepts all search requests and checks if they contain a hybrid query. | ||
| * If a hybrid query is detected with search_type=dfs_query_then_fetch, the request is rejected. | ||
| * If a hybrid query is detected but no search pipeline can be resolved for it (neither inline, | ||
| * via the search_pipeline request parameter, nor as every target index's default search | ||
| * pipeline), the request is rejected, since without a pipeline the normalization processor | ||
| * never runs and the response would otherwise be returned in an unnormalized, internal format. | ||
| * Otherwise, it unconditionally sets batchedReduceSize to Integer.MAX_VALUE to disable batched | ||
| * reduction, regardless of any user-specified value. | ||
| * | ||
| * This prevents the "topDocs already consumed" error that occurs when: | ||
| * 1. Hybrid query is executed | ||
|
|
@@ -40,8 +50,6 @@ | |
| * The NormalizationProcessor requires access to all shard results simultaneously | ||
| * to perform score normalization and combination. | ||
| * | ||
| */ | ||
| @Log4j2 | ||
| public class HybridQuerySearchRequestFilter implements ActionFilter { | ||
|
|
@@ -82,6 +90,10 @@ public <Request extends org.opensearch.action.ActionRequest, Response extends Ac | |
| listener.onFailure(new IllegalArgumentException(HybridQueryUtil.HYBRID_QUERY_DFS_SEARCH_TYPE_NOT_SUPPORTED_MESSAGE)); | ||
| return; | ||
| } | ||
| if (hasResolvableSearchPipeline(searchRequest) == false) { | ||
| listener.onFailure(new IllegalArgumentException(HybridQueryUtil.HYBRID_QUERY_REQUIRES_SEARCH_PIPELINE_MESSAGE)); | ||
| return; | ||
| } | ||
| if (searchRequest.getBatchedReduceSize() != DISABLE_BATCHED_REDUCE) { | ||
| log.debug( | ||
| String.format( | ||
|
|
@@ -119,4 +131,33 @@ private boolean containsHybridQuery(SearchRequest searchRequest) { | |
| // direct check for HybridQueryBuilder | ||
| return query instanceof HybridQueryBuilder; | ||
| } | ||
|
|
||
| /** | ||
| * Check whether a search pipeline can be resolved for this request, either from the request | ||
| * itself (inline or via the search_pipeline request parameter) or from the default search | ||
| * pipeline configured on every index the request targets. | ||
| * | ||
| * @param searchRequest the search request to check | ||
| * @return true if a non-noop pipeline can be resolved for this request | ||
| */ | ||
| private boolean hasResolvableSearchPipeline(SearchRequest searchRequest) { | ||
|
Collaborator
There was a problem hiding this comment. I think we should inspect the pipeline to confirm the normalization and combination processors are configured otherwise a search pipeline without those processors still cannot return the right hybrid query result.
Member
There was a problem hiding this comment. +1 here. Please add a validation to check the normalization and combination processors attached and an associated test
Member
There was a problem hiding this comment. this logic diverges from SearchPipelineService. Core method When Core resolves an inline-body ad-hoc pipeline, request's Example of failing request, where index has no default pipeline, no POST /my-index/_search
{
"search_pipeline": {
"phase_results_processors": [
{ "normalization-processor": {
"normalization": { "technique": "min_max" },
"combination": { "technique": "arithmetic_mean" } } }
]
},
"query": { "hybrid": { "queries": [ {"match":{"text":"hello"}}, {"term":{"text":"place"}} ] } }
}You can fail fast on the inline body before the named/default checks, mirroring core branch. |
||
| String requestPipeline = searchRequest.pipeline(); | ||
| if (isConfiguredPipeline(requestPipeline)) { | ||
| return true; | ||
| } | ||
|
|
||
| List<IndexMetadata> indexMetadataList = NeuralSearchClusterUtil.instance().getIndexMetadataList(searchRequest); | ||
| return indexMetadataList.isEmpty() == false && indexMetadataList.stream().allMatch(this::hasDefaultSearchPipeline); | ||
|
Member
There was a problem hiding this comment. this logic is different in core, and fails in multiple scenarios:
I suggest you don't |
||
| } | ||
|
|
||
| private boolean hasDefaultSearchPipeline(IndexMetadata indexMetadata) { | ||
| String defaultPipeline = IndexSettings.DEFAULT_SEARCH_PIPELINE.get(indexMetadata.getSettings()); | ||
| return isConfiguredPipeline(defaultPipeline); | ||
| } | ||
|
|
||
| private boolean isConfiguredPipeline(String pipelineId) { | ||
| return Objects.nonNull(pipelineId) | ||
| && pipelineId.isBlank() == false | ||
| && SearchPipelineService.NOOP_PIPELINE_ID.equals(pipelineId) == false; | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
provided error message is over promising - we are checking for a simple search pipeline presence, not normalization related processors. Either add check for processor or relax the message and drop mentions of processors (which is much easier fix I believe)