diff --git a/CHANGELOG.md b/CHANGELOG.md index ba9a41b16a37a..9d470c334576b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), - Print reason why parent task was cancelled ([#14604](https://github.com/opensearch-project/OpenSearch/issues/14604)) - Add matchesPluginSystemIndexPattern to SystemIndexRegistry ([#14750](https://github.com/opensearch-project/OpenSearch/pull/14750)) - Add Plugin interface for loading application based configuration templates (([#14659](https://github.com/opensearch-project/OpenSearch/issues/14659))) +- Refactor remote-routing-table service inline with remote state interfaces([#14668](https://github.com/opensearch-project/OpenSearch/pull/14668)) +- Add SortResponseProcessor to Search Pipelines (([#14785](https://github.com/opensearch-project/OpenSearch/issues/14785))) - Add prefix mode verification setting for repository verification (([#14790](https://github.com/opensearch-project/OpenSearch/pull/14790))) - Add SplitResponseProcessor to Search Pipelines (([#14800](https://github.com/opensearch-project/OpenSearch/issues/14800))) - Optimize TransportNodesAction to not send DiscoveryNodes for NodeStats, NodesInfo and ClusterStats call ([14749](https://github.com/opensearch-project/OpenSearch/pull/14749)) diff --git a/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/SearchPipelineCommonModulePlugin.java b/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/SearchPipelineCommonModulePlugin.java index d05101da2817c..2a2de9debb9d9 100644 --- a/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/SearchPipelineCommonModulePlugin.java +++ b/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/SearchPipelineCommonModulePlugin.java @@ -97,8 +97,8 @@ public Map> getResponseProces new TruncateHitsResponseProcessor.Factory(), CollapseResponseProcessor.TYPE, new CollapseResponseProcessor.Factory(), - SplitResponseProcessor.TYPE, - new SplitResponseProcessor.Factory() + SortResponseProcessor.TYPE, + new SortResponseProcessor.Factory() ) ); } diff --git a/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/SortResponseProcessor.java b/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/SortResponseProcessor.java new file mode 100644 index 0000000000000..e0bfd38b26376 --- /dev/null +++ b/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/SortResponseProcessor.java @@ -0,0 +1,209 @@ +/* + * 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.search.pipeline.common; + +import org.opensearch.action.search.SearchRequest; +import org.opensearch.action.search.SearchResponse; +import org.opensearch.common.collect.Tuple; +import org.opensearch.common.document.DocumentField; +import org.opensearch.common.xcontent.XContentHelper; +import org.opensearch.core.common.bytes.BytesReference; +import org.opensearch.core.xcontent.MediaType; +import org.opensearch.core.xcontent.XContentBuilder; +import org.opensearch.ingest.ConfigurationUtils; +import org.opensearch.search.SearchHit; +import org.opensearch.search.pipeline.AbstractProcessor; +import org.opensearch.search.pipeline.Processor; +import org.opensearch.search.pipeline.SearchResponseProcessor; + +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.stream.Collectors; + +/** + * Processor that sorts an array of items. + * Throws exception is the specified field is not an array. + */ +public class SortResponseProcessor extends AbstractProcessor implements SearchResponseProcessor { + /** Key to reference this processor type from a search pipeline. */ + public static final String TYPE = "sort"; + /** Key defining the array field to be sorted. */ + public static final String SORT_FIELD = "field"; + /** Optional key defining the sort order. */ + public static final String SORT_ORDER = "order"; + /** Optional key to put the sorted values in a different field. */ + public static final String TARGET_FIELD = "target_field"; + /** Default sort order if not specified */ + public static final String DEFAULT_ORDER = "asc"; + + /** Enum defining how elements will be sorted */ + public enum SortOrder { + /** Sort in ascending (natural) order */ + ASCENDING("asc"), + /** Sort in descending (reverse) order */ + DESCENDING("desc"); + + private final String direction; + + SortOrder(String direction) { + this.direction = direction; + } + + @Override + public String toString() { + return this.direction; + } + + /** + * Converts the string representation of the enum value to the enum. + * @param value A string ("asc" or "desc") + * @return the corresponding enum value + */ + public static SortOrder fromString(String value) { + if (value == null) { + throw new IllegalArgumentException("Sort direction cannot be null"); + } + + if (value.equals(ASCENDING.toString())) { + return ASCENDING; + } else if (value.equals(DESCENDING.toString())) { + return DESCENDING; + } + throw new IllegalArgumentException("Sort direction [" + value + "] not recognized." + " Valid values are: [asc, desc]"); + } + } + + private final String sortField; + private final SortOrder sortOrder; + private final String targetField; + + SortResponseProcessor( + String tag, + String description, + boolean ignoreFailure, + String sortField, + SortOrder sortOrder, + String targetField + ) { + super(tag, description, ignoreFailure); + this.sortField = Objects.requireNonNull(sortField); + this.sortOrder = Objects.requireNonNull(sortOrder); + this.targetField = targetField == null ? sortField : targetField; + } + + /** + * Getter function for sortField + * @return sortField + */ + public String getSortField() { + return sortField; + } + + /** + * Getter function for targetField + * @return targetField + */ + public String getTargetField() { + return targetField; + } + + /** + * Getter function for sortOrder + * @return sortOrder + */ + public SortOrder getSortOrder() { + return sortOrder; + } + + @Override + public String getType() { + return TYPE; + } + + @Override + public SearchResponse processResponse(SearchRequest request, SearchResponse response) throws Exception { + SearchHit[] hits = response.getHits().getHits(); + for (SearchHit hit : hits) { + Map fields = hit.getFields(); + if (fields.containsKey(sortField)) { + DocumentField docField = hit.getFields().get(sortField); + if (docField == null) { + throw new IllegalArgumentException("field [" + sortField + "] is null, cannot sort."); + } + hit.setDocumentField(targetField, new DocumentField(targetField, getSortedValues(docField.getValues()))); + } + if (hit.hasSource()) { + BytesReference sourceRef = hit.getSourceRef(); + Tuple> typeAndSourceMap = XContentHelper.convertToMap( + sourceRef, + false, + (MediaType) null + ); + + Map sourceAsMap = typeAndSourceMap.v2(); + if (sourceAsMap.containsKey(sortField)) { + Object val = sourceAsMap.get(sortField); + if (val instanceof List) { + @SuppressWarnings("unchecked") + List listVal = (List) val; + sourceAsMap.put(targetField, getSortedValues(listVal)); + } + XContentBuilder builder = XContentBuilder.builder(typeAndSourceMap.v1().xContent()); + builder.map(sourceAsMap); + hit.sourceRef(BytesReference.bytes(builder)); + } + } + } + return response; + } + + private List getSortedValues(List values) { + return values.stream() + .map(this::downcastToComparable) + .sorted(sortOrder.equals(SortOrder.ASCENDING) ? Comparator.naturalOrder() : Comparator.reverseOrder()) + .collect(Collectors.toList()); + } + + @SuppressWarnings("unchecked") + private Comparable downcastToComparable(Object obj) { + if (obj instanceof Comparable) { + return (Comparable) obj; + } else if (obj == null) { + throw new IllegalArgumentException("field [" + sortField + "] contains a null value.]"); + } else { + throw new IllegalArgumentException("field [" + sortField + "] of type [" + obj.getClass().getName() + "] is not comparable.]"); + } + } + + static class Factory implements Processor.Factory { + + @Override + public SortResponseProcessor create( + Map> processorFactories, + String tag, + String description, + boolean ignoreFailure, + Map config, + PipelineContext pipelineContext + ) { + String sortField = ConfigurationUtils.readStringProperty(TYPE, tag, config, SORT_FIELD); + String targetField = ConfigurationUtils.readStringProperty(TYPE, tag, config, TARGET_FIELD, sortField); + try { + SortOrder sortOrder = SortOrder.fromString( + ConfigurationUtils.readStringProperty(TYPE, tag, config, SORT_ORDER, DEFAULT_ORDER) + ); + return new SortResponseProcessor(tag, description, ignoreFailure, sortField, sortOrder, targetField); + } catch (IllegalArgumentException e) { + throw ConfigurationUtils.newConfigurationException(TYPE, tag, SORT_ORDER, e.getMessage()); + } + } + } +} diff --git a/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/SplitResponseProcessor.java b/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/SplitResponseProcessor.java index 0762f8f59b76e..bb3db4d9bc2c1 100644 --- a/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/SplitResponseProcessor.java +++ b/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/SplitResponseProcessor.java @@ -111,7 +111,7 @@ public SearchResponse processResponse(SearchRequest request, SearchResponse resp throw new IllegalArgumentException("field [" + splitField + "] is null, cannot split."); } Object val = docField.getValue(); - if (val == null || !String.class.isAssignableFrom(val.getClass())) { + if (!(val instanceof String)) { throw new IllegalArgumentException("field [" + splitField + "] is not a string, cannot split"); } Object[] strings = ((String) val).split(separator, preserveTrailing ? -1 : 0); diff --git a/modules/search-pipeline-common/src/test/java/org/opensearch/search/pipeline/common/SearchPipelineCommonModulePluginTests.java b/modules/search-pipeline-common/src/test/java/org/opensearch/search/pipeline/common/SearchPipelineCommonModulePluginTests.java index d4f9ae2490a10..404842742629c 100644 --- a/modules/search-pipeline-common/src/test/java/org/opensearch/search/pipeline/common/SearchPipelineCommonModulePluginTests.java +++ b/modules/search-pipeline-common/src/test/java/org/opensearch/search/pipeline/common/SearchPipelineCommonModulePluginTests.java @@ -82,7 +82,7 @@ public void testAllowlistNotSpecified() throws IOException { try (SearchPipelineCommonModulePlugin plugin = new SearchPipelineCommonModulePlugin()) { assertEquals(Set.of("oversample", "filter_query", "script"), plugin.getRequestProcessors(createParameters(settings)).keySet()); assertEquals( - Set.of("rename_field", "truncate_hits", "collapse", "split"), + Set.of("rename_field", "truncate_hits", "collapse", "sort"), plugin.getResponseProcessors(createParameters(settings)).keySet() ); assertEquals(Set.of(), plugin.getSearchPhaseResultsProcessors(createParameters(settings)).keySet()); diff --git a/modules/search-pipeline-common/src/test/java/org/opensearch/search/pipeline/common/SortResponseProcessorTests.java b/modules/search-pipeline-common/src/test/java/org/opensearch/search/pipeline/common/SortResponseProcessorTests.java new file mode 100644 index 0000000000000..c18c6b34b05d1 --- /dev/null +++ b/modules/search-pipeline-common/src/test/java/org/opensearch/search/pipeline/common/SortResponseProcessorTests.java @@ -0,0 +1,230 @@ +/* + * 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.java + * compatible open source license. + */ + +package org.opensearch.search.pipeline.common; + +import org.apache.lucene.search.TotalHits; +import org.opensearch.OpenSearchParseException; +import org.opensearch.action.search.SearchRequest; +import org.opensearch.action.search.SearchResponse; +import org.opensearch.action.search.SearchResponseSections; +import org.opensearch.common.document.DocumentField; +import org.opensearch.core.common.bytes.BytesArray; +import org.opensearch.index.query.QueryBuilder; +import org.opensearch.index.query.TermQueryBuilder; +import org.opensearch.ingest.RandomDocumentPicks; +import org.opensearch.search.SearchHit; +import org.opensearch.search.SearchHits; +import org.opensearch.search.builder.SearchSourceBuilder; +import org.opensearch.test.OpenSearchTestCase; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public class SortResponseProcessorTests extends OpenSearchTestCase { + + private static final List PI = List.of(3, 1, 4, 1, 5, 9, 2, 6); + private static final List E = List.of(2, 7, 1, 8, 2, 8, 1, 8); + private static final List X; + static { + List x = new ArrayList<>(); + x.add(1); + x.add(null); + x.add(3); + X = x; + } + + private SearchRequest createDummyRequest() { + QueryBuilder query = new TermQueryBuilder("field", "value"); + SearchSourceBuilder source = new SearchSourceBuilder().query(query); + return new SearchRequest().source(source); + } + + private SearchResponse createTestResponse() { + SearchHit[] hits = new SearchHit[2]; + + // one response with source + Map piMap = new HashMap<>(); + piMap.put("digits", new DocumentField("digits", PI)); + hits[0] = new SearchHit(0, "doc 1", piMap, Collections.emptyMap()); + hits[0].sourceRef(new BytesArray("{ \"digits\" : " + PI + " }")); + hits[0].score((float) Math.PI); + + // one without source + Map eMap = new HashMap<>(); + eMap.put("digits", new DocumentField("digits", E)); + hits[1] = new SearchHit(1, "doc 2", eMap, Collections.emptyMap()); + hits[1].score((float) Math.E); + + SearchHits searchHits = new SearchHits(hits, new TotalHits(2, TotalHits.Relation.EQUAL_TO), 2); + SearchResponseSections searchResponseSections = new SearchResponseSections(searchHits, null, null, false, false, null, 0); + return new SearchResponse(searchResponseSections, null, 1, 1, 0, 10, null, null); + } + + private SearchResponse createTestResponseNullField() { + SearchHit[] hits = new SearchHit[1]; + + Map map = new HashMap<>(); + map.put("digits", null); + hits[0] = new SearchHit(0, "doc 1", map, Collections.emptyMap()); + hits[0].sourceRef(new BytesArray("{ \"digits\" : null }")); + hits[0].score((float) Math.PI); + + SearchHits searchHits = new SearchHits(hits, new TotalHits(1, TotalHits.Relation.EQUAL_TO), 1); + SearchResponseSections searchResponseSections = new SearchResponseSections(searchHits, null, null, false, false, null, 0); + return new SearchResponse(searchResponseSections, null, 1, 1, 0, 10, null, null); + } + + private SearchResponse createTestResponseNullListEntry() { + SearchHit[] hits = new SearchHit[1]; + + Map xMap = new HashMap<>(); + xMap.put("digits", new DocumentField("digits", X)); + hits[0] = new SearchHit(0, "doc 1", xMap, Collections.emptyMap()); + hits[0].sourceRef(new BytesArray("{ \"digits\" : " + X + " }")); + hits[0].score((float) Math.PI); + + SearchHits searchHits = new SearchHits(hits, new TotalHits(1, TotalHits.Relation.EQUAL_TO), 1); + SearchResponseSections searchResponseSections = new SearchResponseSections(searchHits, null, null, false, false, null, 0); + return new SearchResponse(searchResponseSections, null, 1, 1, 0, 10, null, null); + } + + private SearchResponse createTestResponseNotComparable() { + SearchHit[] hits = new SearchHit[1]; + + Map piMap = new HashMap<>(); + piMap.put("maps", new DocumentField("maps", List.of(Map.of("foo", "I'm incomparable!")))); + hits[0] = new SearchHit(0, "doc 1", piMap, Collections.emptyMap()); + hits[0].sourceRef(new BytesArray("{ \"maps\" : [{ \"foo\" : \"I'm incomparable!\"}]] }")); + hits[0].score((float) Math.PI); + + SearchHits searchHits = new SearchHits(hits, new TotalHits(1, TotalHits.Relation.EQUAL_TO), 1); + SearchResponseSections searchResponseSections = new SearchResponseSections(searchHits, null, null, false, false, null, 0); + return new SearchResponse(searchResponseSections, null, 1, 1, 0, 10, null, null); + } + + public void testSortResponse() throws Exception { + SearchRequest request = createDummyRequest(); + + SortResponseProcessor sortResponseProcessor = new SortResponseProcessor( + null, + null, + false, + "digits", + SortResponseProcessor.SortOrder.ASCENDING, + "sorted" + ); + SearchResponse response = createTestResponse(); + SearchResponse sortResponse = sortResponseProcessor.processResponse(request, response); + + assertEquals(response.getHits(), sortResponse.getHits()); + + assertEquals(PI, sortResponse.getHits().getHits()[0].field("digits").getValues()); + assertEquals(List.of(1, 1, 2, 3, 4, 5, 6, 9), sortResponse.getHits().getHits()[0].field("sorted").getValues()); + Map map = sortResponse.getHits().getHits()[0].getSourceAsMap(); + assertNotNull(map); + assertEquals(List.of(1, 1, 2, 3, 4, 5, 6, 9), map.get("sorted")); + + assertEquals(E, sortResponse.getHits().getHits()[1].field("digits").getValues()); + assertEquals(List.of(1, 1, 2, 2, 7, 8, 8, 8), sortResponse.getHits().getHits()[1].field("sorted").getValues()); + assertNull(sortResponse.getHits().getHits()[1].getSourceAsMap()); + } + + public void testSortResponseSameField() throws Exception { + SearchRequest request = createDummyRequest(); + + SortResponseProcessor sortResponseProcessor = new SortResponseProcessor( + null, + null, + false, + "digits", + SortResponseProcessor.SortOrder.DESCENDING, + null + ); + SearchResponse response = createTestResponse(); + SearchResponse sortResponse = sortResponseProcessor.processResponse(request, response); + + assertEquals(response.getHits(), sortResponse.getHits()); + assertEquals(List.of(9, 6, 5, 4, 3, 2, 1, 1), sortResponse.getHits().getHits()[0].field("digits").getValues()); + assertEquals(List.of(8, 8, 8, 7, 2, 2, 1, 1), sortResponse.getHits().getHits()[1].field("digits").getValues()); + } + + public void testSortResponseNullListEntry() { + SearchRequest request = createDummyRequest(); + + SortResponseProcessor sortResponseProcessor = new SortResponseProcessor( + null, + null, + false, + "digits", + SortResponseProcessor.SortOrder.ASCENDING, + null + ); + assertThrows( + IllegalArgumentException.class, + () -> sortResponseProcessor.processResponse(request, createTestResponseNullListEntry()) + ); + } + + public void testNullField() { + SearchRequest request = createDummyRequest(); + + SortResponseProcessor sortResponseProcessor = new SortResponseProcessor( + null, + null, + false, + "digits", + SortResponseProcessor.SortOrder.DESCENDING, + null + ); + + assertThrows(IllegalArgumentException.class, () -> sortResponseProcessor.processResponse(request, createTestResponseNullField())); + } + + public void testNotComparableField() { + SearchRequest request = createDummyRequest(); + + SortResponseProcessor sortResponseProcessor = new SortResponseProcessor( + null, + null, + false, + "maps", + SortResponseProcessor.SortOrder.ASCENDING, + null + ); + + assertThrows( + IllegalArgumentException.class, + () -> sortResponseProcessor.processResponse(request, createTestResponseNotComparable()) + ); + } + + public void testFactory() { + String sortField = RandomDocumentPicks.randomFieldName(random()); + String targetField = RandomDocumentPicks.randomFieldName(random()); + Map config = new HashMap<>(); + config.put("field", sortField); + config.put("order", "desc"); + config.put("target_field", targetField); + + SortResponseProcessor.Factory factory = new SortResponseProcessor.Factory(); + SortResponseProcessor processor = factory.create(Collections.emptyMap(), null, null, false, config, null); + assertEquals("sort", processor.getType()); + assertEquals(sortField, processor.getSortField()); + assertEquals(targetField, processor.getTargetField()); + assertEquals(SortResponseProcessor.SortOrder.DESCENDING, processor.getSortOrder()); + + expectThrows( + OpenSearchParseException.class, + () -> factory.create(Collections.emptyMap(), null, null, false, Collections.emptyMap(), null) + ); + } +} diff --git a/modules/search-pipeline-common/src/yamlRestTest/resources/rest-api-spec/test/search_pipeline/80_sort_response.yml b/modules/search-pipeline-common/src/yamlRestTest/resources/rest-api-spec/test/search_pipeline/80_sort_response.yml new file mode 100644 index 0000000000000..c160b550b2a6e --- /dev/null +++ b/modules/search-pipeline-common/src/yamlRestTest/resources/rest-api-spec/test/search_pipeline/80_sort_response.yml @@ -0,0 +1,152 @@ +--- +teardown: + - do: + search_pipeline.delete: + id: "my_pipeline" + ignore: 404 + +--- +"Test sort processor": + - do: + search_pipeline.put: + id: "my_pipeline" + body: > + { + "description": "test pipeline", + "response_processors": [ + { + "sort": + { + "field": "a", + "target_field": "b" + } + } + ] + } + - match: { acknowledged: true } + + - do: + search_pipeline.put: + id: "my_pipeline_2" + body: > + { + "description": "test pipeline with ignore failure true", + "response_processors": [ + { + "sort": + { + "field": "aa", + "ignore_failure": true + } + } + ] + } + - match: { acknowledged: true } + + - do: + search_pipeline.put: + id: "my_pipeline_3" + body: > + { + "description": "test pipeline", + "response_processors": [ + { + "sort": + { + "field": "a", + "order": "desc", + "target_field": "b" + } + } + ] + } + - match: { acknowledged: true } + + - do: + indices.create: + index: test + + - do: + indices.put_mapping: + index: test + body: + properties: + a: + type: integer + store: true + doc_values: true + + - do: + index: + index: test + id: 1 + body: { + "a": [ 3, 1, 4 ] + } + + - do: + indices.refresh: + index: test + + - do: + search: + body: { } + - match: { hits.total.value: 1 } + + - do: + search: + index: test + search_pipeline: "my_pipeline" + body: { } + - match: { hits.total.value: 1 } + - match: { hits.hits.0._source: { "a": [3, 1, 4], "b": [1, 3, 4] } } + + # Should also work with no search body specified + - do: + search: + index: test + search_pipeline: "my_pipeline" + - match: { hits.total.value: 1 } + - match: { hits.hits.0._source: { "a": [3, 1, 4], "b": [1, 3, 4] } } + + # Pipeline with ignore_failure set to true + # Should return while catching error + - do: + search: + index: test + search_pipeline: "my_pipeline_2" + - match: { hits.total.value: 1 } + - match: { hits.hits.0._source: { "a": [3, 1, 4] } } + + # Pipeline with desc sort order + - do: + search: + index: test + search_pipeline: "my_pipeline_3" + body: { } + - match: { hits.total.value: 1 } + - match: { hits.hits.0._source: { "a": [3, 1, 4], "b": [4, 3, 1] } } + + # No source, using stored_fields + - do: + search: + index: test + search_pipeline: "my_pipeline" + body: { + "_source": false, + "stored_fields": [ "a" ] + } + - match: { hits.hits.0.fields: { "a": [3, 1, 4], "b": [1, 3, 4] } } + + # No source, using docvalue_fields + - do: + search: + index: test + search_pipeline: "my_pipeline_3" + body: { + "_source": false, + "docvalue_fields": [ "a" ] + } + # a is stored sorted because docvalue_fields is pre-sorted to optimize aggregations + # this is poorly documented which makes it really hard to write "expected" values on tests + - match: { hits.hits.0.fields: { "a": [1, 3, 4], "b": [4, 3, 1] } }