diff --git a/server/src/main/java/org/opensearch/action/search/AbstractSearchAsyncAction.java b/server/src/main/java/org/opensearch/action/search/AbstractSearchAsyncAction.java index 26d78caa7bd4c..5c03a12984aee 100644 --- a/server/src/main/java/org/opensearch/action/search/AbstractSearchAsyncAction.java +++ b/server/src/main/java/org/opensearch/action/search/AbstractSearchAsyncAction.java @@ -699,7 +699,7 @@ private void raisePhaseFailure(SearchPhaseExecutionException exception) { final void onPhaseDone() { // as a tribute to @kimchy aka. finishHim() final SearchPhase nextPhase = getNextPhase(results, this); if (request instanceof PipelinedRequest && nextPhase != null) { - ((PipelinedRequest) request).transformSearchPhase(results, this, this.getName(), nextPhase.getName()); + ((PipelinedRequest) request).transformSearchPhaseResults(results, this, this.getName(), nextPhase.getName()); } executeNextPhase(this, nextPhase); } diff --git a/server/src/main/java/org/opensearch/action/search/SearchPhase.java b/server/src/main/java/org/opensearch/action/search/SearchPhase.java index 43dbbc18a30db..50b0cd8e01c1d 100644 --- a/server/src/main/java/org/opensearch/action/search/SearchPhase.java +++ b/server/src/main/java/org/opensearch/action/search/SearchPhase.java @@ -42,7 +42,7 @@ * * @opensearch.internal */ -public abstract class SearchPhase implements CheckedRunnable { +abstract class SearchPhase implements CheckedRunnable { private final String name; protected SearchPhase(String name) { @@ -64,26 +64,4 @@ public String getName() { public SearchPhaseName getSearchPhaseName() { return SearchPhaseName.valueOf(name.toUpperCase(Locale.ROOT)); } - - /** - * Enum for different Search Phases in OpenSearch - * @opensearch.internal - */ - public enum SearchPhaseName { - QUERY("query"), - FETCH("fetch"), - DFS_QUERY("dfs_query"), - EXPAND("expand"), - CAN_MATCH("can_match"); - - private final String name; - - SearchPhaseName(final String name) { - this.name = name; - } - - public String getName() { - return name; - } - } } diff --git a/server/src/main/java/org/opensearch/action/search/SearchPhaseName.java b/server/src/main/java/org/opensearch/action/search/SearchPhaseName.java new file mode 100644 index 0000000000000..b6f842cf2cce1 --- /dev/null +++ b/server/src/main/java/org/opensearch/action/search/SearchPhaseName.java @@ -0,0 +1,31 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.action.search; + +/** + * Enum for different Search Phases in OpenSearch + * @opensearch.internal + */ +public enum SearchPhaseName { + QUERY("query"), + FETCH("fetch"), + DFS_QUERY("dfs_query"), + EXPAND("expand"), + CAN_MATCH("can_match"); + + private final String name; + + SearchPhaseName(final String name) { + this.name = name; + } + + public String getName() { + return name; + } +} diff --git a/server/src/main/java/org/opensearch/action/search/SearchScrollAsyncAction.java b/server/src/main/java/org/opensearch/action/search/SearchScrollAsyncAction.java index a0164971f3d19..899c7a3c1dabd 100644 --- a/server/src/main/java/org/opensearch/action/search/SearchScrollAsyncAction.java +++ b/server/src/main/java/org/opensearch/action/search/SearchScrollAsyncAction.java @@ -266,7 +266,7 @@ protected SearchPhase sendResponsePhase( SearchPhaseController.ReducedQueryPhase queryPhase, final AtomicArray fetchResults ) { - return new SearchPhase(SearchPhase.SearchPhaseName.FETCH.getName()) { + return new SearchPhase(SearchPhaseName.FETCH.getName()) { @Override public void run() throws IOException { sendResponse(queryPhase, fetchResults); diff --git a/server/src/main/java/org/opensearch/action/search/SearchScrollQueryThenFetchAsyncAction.java b/server/src/main/java/org/opensearch/action/search/SearchScrollQueryThenFetchAsyncAction.java index 51ffeb2ac83bc..9c0721ef63ea6 100644 --- a/server/src/main/java/org/opensearch/action/search/SearchScrollQueryThenFetchAsyncAction.java +++ b/server/src/main/java/org/opensearch/action/search/SearchScrollQueryThenFetchAsyncAction.java @@ -92,7 +92,7 @@ protected void executeInitialPhase( @Override protected SearchPhase moveToNextPhase(BiFunction clusterNodeLookup) { - return new SearchPhase(SearchPhase.SearchPhaseName.FETCH.getName()) { + return new SearchPhase(SearchPhaseName.FETCH.getName()) { @Override public void run() { final SearchPhaseController.ReducedQueryPhase reducedQueryPhase = searchPhaseController.reducedScrollQueryPhase( diff --git a/server/src/main/java/org/opensearch/plugins/SearchPipelinePlugin.java b/server/src/main/java/org/opensearch/plugins/SearchPipelinePlugin.java index 948b7790d56bd..3d76bab93a60c 100644 --- a/server/src/main/java/org/opensearch/plugins/SearchPipelinePlugin.java +++ b/server/src/main/java/org/opensearch/plugins/SearchPipelinePlugin.java @@ -51,7 +51,7 @@ default Map> getResponseProce * in pipeline configurations, and the value is a {@link org.opensearch.search.pipeline.Processor.Factory} * to create the processor from a given pipeline configuration. */ - default Map> getPhaseResultsProcessors(Processor.Parameters parameters) { + default Map> getSearchPhaseResultsProcessors(Processor.Parameters parameters) { return Collections.emptyMap(); } } diff --git a/server/src/main/java/org/opensearch/search/pipeline/Pipeline.java b/server/src/main/java/org/opensearch/search/pipeline/Pipeline.java index 9162cf97ceda5..12cc8f14338c5 100644 --- a/server/src/main/java/org/opensearch/search/pipeline/Pipeline.java +++ b/server/src/main/java/org/opensearch/search/pipeline/Pipeline.java @@ -203,7 +203,7 @@ SearchResponse transformResponse(SearchRequest request, SearchResponse response) null ); - void runSearchPhaseTransformer( + void runSearchPhaseResultsTransformer( SearchPhaseResults searchPhaseResult, SearchPhaseContext context, String currentPhase, diff --git a/server/src/main/java/org/opensearch/search/pipeline/PipelinedRequest.java b/server/src/main/java/org/opensearch/search/pipeline/PipelinedRequest.java index 8d91876f34f11..5a7539808c127 100644 --- a/server/src/main/java/org/opensearch/search/pipeline/PipelinedRequest.java +++ b/server/src/main/java/org/opensearch/search/pipeline/PipelinedRequest.java @@ -31,13 +31,13 @@ public SearchResponse transformResponse(SearchResponse response) { return pipeline.transformResponse(this, response); } - public void transformSearchPhase( + public void transformSearchPhaseResults( final SearchPhaseResults searchPhaseResult, final SearchPhaseContext searchPhaseContext, final String currentPhase, final String nextPhase ) { - pipeline.runSearchPhaseTransformer(searchPhaseResult, searchPhaseContext, currentPhase, nextPhase); + pipeline.runSearchPhaseResultsTransformer(searchPhaseResult, searchPhaseContext, currentPhase, nextPhase); } // Visible for testing diff --git a/server/src/main/java/org/opensearch/search/pipeline/SearchPhaseResultsProcessor.java b/server/src/main/java/org/opensearch/search/pipeline/SearchPhaseResultsProcessor.java index 783a69114f50f..772dc8758bace 100644 --- a/server/src/main/java/org/opensearch/search/pipeline/SearchPhaseResultsProcessor.java +++ b/server/src/main/java/org/opensearch/search/pipeline/SearchPhaseResultsProcessor.java @@ -8,8 +8,8 @@ package org.opensearch.search.pipeline; -import org.opensearch.action.search.SearchPhase; import org.opensearch.action.search.SearchPhaseContext; +import org.opensearch.action.search.SearchPhaseName; import org.opensearch.action.search.SearchPhaseResults; import org.opensearch.search.SearchPhaseResult; import org.opensearch.search.internal.SearchContext; @@ -21,8 +21,8 @@ public interface SearchPhaseResultsProcessor extends Processor { /** - * Processes the {@link SearchPhaseResults} obtained from a {@link SearchPhase} which will be returned to next - * {@link SearchPhase}. + * Processes the {@link SearchPhaseResults} obtained from a SearchPhase which will be returned to next + * SearchPhase. * @param searchPhaseResult {@link SearchPhaseResults} * @param searchPhaseContext {@link SearchContext} * @param {@link SearchPhaseResult} @@ -34,14 +34,14 @@ void process( /** * The phase which should have run before, this processor can start executing. - * @return {@link SearchPhase.SearchPhaseName} + * @return {@link SearchPhaseName} */ - SearchPhase.SearchPhaseName getBeforePhase(); + SearchPhaseName getBeforePhase(); /** * The phase which should run after, this processor execution. - * @return {@link SearchPhase.SearchPhaseName} + * @return {@link SearchPhaseName} */ - SearchPhase.SearchPhaseName getAfterPhase(); + SearchPhaseName getAfterPhase(); } diff --git a/server/src/main/java/org/opensearch/search/pipeline/SearchPipelineService.java b/server/src/main/java/org/opensearch/search/pipeline/SearchPipelineService.java index 1dbc2d0609cfc..29bab3aac6910 100644 --- a/server/src/main/java/org/opensearch/search/pipeline/SearchPipelineService.java +++ b/server/src/main/java/org/opensearch/search/pipeline/SearchPipelineService.java @@ -114,7 +114,10 @@ public SearchPipelineService( ); this.requestProcessorFactories = processorFactories(searchPipelinePlugins, p -> p.getRequestProcessors(parameters)); this.responseProcessorFactories = processorFactories(searchPipelinePlugins, p -> p.getResponseProcessors(parameters)); - this.phaseInjectorProcessorFactories = processorFactories(searchPipelinePlugins, p -> p.getPhaseResultsProcessors(parameters)); + this.phaseInjectorProcessorFactories = processorFactories( + searchPipelinePlugins, + p -> p.getSearchPhaseResultsProcessors(parameters) + ); putPipelineTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTaskKeys.PUT_SEARCH_PIPELINE_KEY, true); deletePipelineTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTaskKeys.DELETE_SEARCH_PIPELINE_KEY, true); this.isEnabled = isEnabled; diff --git a/server/src/test/java/org/opensearch/search/pipeline/SearchPipelineServiceTests.java b/server/src/test/java/org/opensearch/search/pipeline/SearchPipelineServiceTests.java index 04bc4ed1d4699..6685245748fef 100644 --- a/server/src/test/java/org/opensearch/search/pipeline/SearchPipelineServiceTests.java +++ b/server/src/test/java/org/opensearch/search/pipeline/SearchPipelineServiceTests.java @@ -21,9 +21,9 @@ import org.opensearch.action.search.MockSearchPhaseContext; import org.opensearch.action.search.PutSearchPipelineRequest; import org.opensearch.action.search.QueryPhaseResultConsumer; -import org.opensearch.action.search.SearchPhase; import org.opensearch.action.search.SearchPhaseContext; import org.opensearch.action.search.SearchPhaseController; +import org.opensearch.action.search.SearchPhaseName; import org.opensearch.action.search.SearchPhaseResults; import org.opensearch.action.search.SearchProgressListener; import org.opensearch.action.search.SearchRequest; @@ -85,7 +85,9 @@ public Map> getResponseProces } @Override - public Map> getPhaseResultsProcessors(Processor.Parameters parameters) { + public Map> getSearchPhaseResultsProcessors( + Processor.Parameters parameters + ) { return Map.of("zoe", (factories, tag, description, config) -> null); } }; @@ -288,13 +290,13 @@ public void process( } @Override - public SearchPhase.SearchPhaseName getBeforePhase() { - return SearchPhase.SearchPhaseName.QUERY; + public SearchPhaseName getBeforePhase() { + return SearchPhaseName.QUERY; } @Override - public SearchPhase.SearchPhaseName getAfterPhase() { - return SearchPhase.SearchPhaseName.FETCH; + public SearchPhaseName getAfterPhase() { + return SearchPhaseName.FETCH; } } @@ -361,7 +363,7 @@ public Map> getResponseProces } @Override - public Map> getPhaseResultsProcessors( + public Map> getSearchPhaseResultsProcessors( Processor.Parameters parameters ) { return phaseProcessors; @@ -683,11 +685,11 @@ public void testTransformSearchPhase() { SearchRequest searchRequest = new SearchRequest(); PipelinedRequest pipelinedRequest = searchPipelineService.resolvePipeline(searchRequest); AtomicArray notTransformedSearchPhaseResults = searchPhaseResults.getAtomicArray(); - pipelinedRequest.transformSearchPhase( + pipelinedRequest.transformSearchPhaseResults( searchPhaseResults, searchPhaseContext, - SearchPhase.SearchPhaseName.QUERY.getName(), - SearchPhase.SearchPhaseName.FETCH.getName() + SearchPhaseName.QUERY.getName(), + SearchPhaseName.FETCH.getName() ); assertSame(searchPhaseResults.getAtomicArray(), notTransformedSearchPhaseResults); @@ -695,11 +697,11 @@ public void testTransformSearchPhase() { searchRequest = new SearchRequest().pipeline("p1"); pipelinedRequest = searchPipelineService.resolvePipeline(searchRequest); - pipelinedRequest.transformSearchPhase( + pipelinedRequest.transformSearchPhaseResults( searchPhaseResults, searchPhaseContext, - SearchPhase.SearchPhaseName.QUERY.getName(), - SearchPhase.SearchPhaseName.FETCH.getName() + SearchPhaseName.QUERY.getName(), + SearchPhaseName.FETCH.getName() ); List resultAtomicArray = searchPhaseResults.getAtomicArray().asList(); @@ -713,11 +715,11 @@ public void testTransformSearchPhase() { searchRequest = new SearchRequest().pipeline("p1"); pipelinedRequest = searchPipelineService.resolvePipeline(searchRequest); AtomicArray notTransformedSearchPhaseResult = searchPhaseResults.getAtomicArray(); - pipelinedRequest.transformSearchPhase( + pipelinedRequest.transformSearchPhaseResults( searchPhaseResults, searchPhaseContext, - SearchPhase.SearchPhaseName.DFS_QUERY.getName(), - SearchPhase.SearchPhaseName.QUERY.getName() + SearchPhaseName.DFS_QUERY.getName(), + SearchPhaseName.QUERY.getName() ); assertSame(searchPhaseResults.getAtomicArray(), notTransformedSearchPhaseResult);