diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLDedupIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLDedupIT.java index 9c93b12e6ac..4177d108440 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLDedupIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLDedupIT.java @@ -348,6 +348,45 @@ public void testMultiColumnSortThenDedup() throws IOException { verifyDataRows(actual, rows("M", "AK", 20, 23), rows("F", "AK", 21, 334)); } + /** + * Regression test for https://github.com/opensearch-project/sql/issues/5150 + * + *

A renamed field that is not the dedup key must retain its value after dedup aggregation + * pushdown. Previously the top_hits response returned the original index field name ({@code + * name}), which the enumerator could not resolve to the renamed output name ({@code nm}), + * yielding null. + */ + @Test + public void testDedupWithRenamedField() throws IOException { + JSONObject actual = + executeQuery( + String.format( + "source=%s | rename name as nm | dedup 1 category | fields category, nm", + TEST_INDEX_DUPLICATION_NULLABLE)); + // One representative row per category; nm must not be null + verifyDataRows(actual, rows("X", "A"), rows("Z", "B"), rows("Y", "A")); + } + + /** + * Regression test for https://github.com/opensearch-project/sql/issues/5197 + * + *

When both a {@code rename} and an {@code eval} column-reference resolve to the same original + * index field, the old {@code Map<String,String>} mapping silently dropped one alias on + * collision. With {@code Map<String,List<String>>} both aliases must appear in the + * result with correct values. + */ + @Test + public void testDedupWithRenamedFieldMappingCollision() throws IOException { + JSONObject actual = + executeQuery( + String.format( + "source=%s | eval nm2 = name | rename name as nm | dedup 1 category" + + " | fields category, nm, nm2", + TEST_INDEX_DUPLICATION_NULLABLE)); + // Both nm (from rename) and nm2 (from eval col-ref) must carry the same non-null name value + verifyDataRows(actual, rows("X", "A", "A"), rows("Z", "B", "B"), rows("Y", "A", "A")); + } + /** Regression test for https://github.com/opensearch-project/sql/issues/3922 */ @Test public void testSortThenDedupKeepEmpty() throws IOException { diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/AggregateAnalyzer.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/AggregateAnalyzer.java index f919fdc0e30..775b0278683 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/AggregateAnalyzer.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/AggregateAnalyzer.java @@ -38,6 +38,7 @@ import com.google.common.collect.ImmutableList; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -618,7 +619,31 @@ yield switch (functionName) { topHitsAggregationBuilder.sort( SortBuilders.fieldSort(key.field()).order(order).missing(missing)); } - yield Pair.of(topHitsAggregationBuilder, new TopHitsParser(aggName, false, false)); + // Build a mapping from original index field name to output (renamed) field names. + // top_hits returns _source / fields keyed by the original index name, but the Calcite + // row-type uses the renamed output name. A single original field may map to multiple + // output names when both a rename and an eval column-ref resolve to the same source + // field (issue #5197), so the value is a List rather than a single String. + Map> fieldNameMapping = new HashMap<>(); + for (Pair arg : args) { + if (arg.getKey() instanceof RexInputRef) { + NamedFieldExpression namedField = helper.inferNamedField(arg.getKey()); + if (namedField == null) { + continue; + } + String originalName = namedField.getRootName(); + String outputName = arg.getValue(); + if (!originalName.equals(outputName)) { + fieldNameMapping + .computeIfAbsent(originalName, k -> new ArrayList<>()) + .add(outputName); + } + } + } + yield Pair.of( + topHitsAggregationBuilder, + new TopHitsParser( + aggName, false, false, fieldNameMapping.isEmpty() ? null : fieldNameMapping)); } default -> throw new AggregateAnalyzer.AggregateAnalyzerException( diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/response/agg/TopHitsParser.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/response/agg/TopHitsParser.java index f9c3d5bb5d2..9fa3589f1fb 100644 --- a/opensearch/src/main/java/org/opensearch/sql/opensearch/response/agg/TopHitsParser.java +++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/response/agg/TopHitsParser.java @@ -12,6 +12,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import javax.annotation.Nullable; import lombok.EqualsAndHashCode; import lombok.Getter; import org.opensearch.common.document.DocumentField; @@ -27,10 +28,29 @@ public class TopHitsParser implements MetricParser { private final boolean returnSingleValue; private final boolean returnMergeValue; + /** + * Maps each original OpenSearch field name to one or more output (renamed) field names. Used by + * the dedup aggregation pushdown path when {@code rename} creates aliases that differ from the + * index field name: top_hits returns {@code _source} / {@code fields} entries keyed by the + * original name, while the Calcite row-type expects the renamed name. A single original field may + * map to multiple output names when both {@code rename} and an {@code eval} column reference + * resolve to the same source field (issue #5197). + */ + @Nullable private final Map> fieldNameMapping; + public TopHitsParser(String name, boolean returnSingleValue, boolean returnMergeValue) { + this(name, returnSingleValue, returnMergeValue, null); + } + + public TopHitsParser( + String name, + boolean returnSingleValue, + boolean returnMergeValue, + @Nullable Map> fieldNameMapping) { this.name = name; this.returnSingleValue = returnSingleValue; this.returnMergeValue = returnMergeValue; + this.fieldNameMapping = fieldNameMapping; } @Override @@ -129,12 +149,43 @@ public List> parse(Aggregation agg) { ? new LinkedHashMap<>() : new LinkedHashMap<>(hit.getSourceAsMap()); hit.getFields().values().forEach(f -> map.put(f.getName(), f.getValue())); + applyFieldNameMapping(map); return map; }) .toList(); } } + /** + * Applies {@link #fieldNameMapping} to a parsed hit map in-place. + * + *

For each {@code (originalName → [outputName1, outputName2, ...])} entry: the value stored + * under {@code originalName} is copied to every output name, and {@code originalName} is removed + * unless it is itself one of the expected output names. This handles both the single-rename case + * (issue #5150) and the many-to-one collision case where two aliases resolve to the same source + * field (issue #5197). + */ + private void applyFieldNameMapping(Map map) { + if (fieldNameMapping == null || fieldNameMapping.isEmpty()) { + return; + } + for (Map.Entry> entry : fieldNameMapping.entrySet()) { + String originalName = entry.getKey(); + Object value = map.get(originalName); + if (value == null && !map.containsKey(originalName)) { + continue; + } + List outputNames = entry.getValue(); + for (String outputName : outputNames) { + map.put(outputName, value); + } + // Remove the original key only when it is not itself one of the expected output names. + if (!outputNames.contains(originalName)) { + map.remove(originalName); + } + } + } + private boolean isEmptyHits(SearchHit[] hits) { return isFieldsEmpty(hits) && isSourceEmpty(hits); } diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/response/OpenSearchAggregationResponseParserTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/response/OpenSearchAggregationResponseParserTest.java index 7ba64eaa475..1147fccf063 100644 --- a/opensearch/src/test/java/org/opensearch/sql/opensearch/response/OpenSearchAggregationResponseParserTest.java +++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/response/OpenSearchAggregationResponseParserTest.java @@ -570,6 +570,96 @@ void two_bucket_percentiles_should_pass() { ImmutableMap.of("percentiles", List.of(21.0, 27.0, 30.0, 35.0, 55.0, 58.0, 60.0)))); } + /** + * Dedup pushdown (LITERAL_AGG) with a renamed field: {@code rename value as val | dedup + * category}. The top_hits response returns {@code _source: {category, value}} but the output + * schema expects {@code val}. The {@code fieldNameMapping} must translate {@code value -> val}. + * + *

Regression test for https://github.com/opensearch-project/sql/issues/5150 + */ + @Test + void top_hits_field_name_mapping_single_rename_should_pass() { + String response = + "{\n" + + " \"composite#composite_buckets\": {\n" + + " \"buckets\": [\n" + + " {\n" + + " \"key\": { \"category\": \"A\" },\n" + + " \"doc_count\": 2,\n" + + " \"top_hits#dedup\": {\n" + + " \"hits\": {\n" + + " \"total\": { \"value\": 2, \"relation\": \"eq\" },\n" + + " \"hits\": [\n" + + " {\n" + + " \"_index\": \"idx\",\n" + + " \"_id\": \"1\",\n" + + " \"fields\": { \"value\": [10.5] }\n" + + " }\n" + + " ]\n" + + " }\n" + + " }\n" + + " }\n" + + " ]\n" + + " }\n" + + "}"; + + Map> mapping = Map.of("value", List.of("val")); + OpenSearchAggregationResponseParser parser = + new CompositeAggregationParser(new TopHitsParser("dedup", false, false, mapping)); + assertThat( + parse(parser, response), + contains(ImmutableMap.of("category", "A"), ImmutableMap.of("val", 10.5))); + } + + /** + * Dedup pushdown (LITERAL_AGG) where two output names ({@code pay} from rename, {@code pay2} from + * eval column-ref) both resolve to the same original field {@code salary}. The old {@code + * Map} approach silently dropped one mapping on collision; with {@code + * Map>} both aliases must appear in the result. + * + *

Regression test for https://github.com/opensearch-project/sql/issues/5197 + */ + @Test + void top_hits_field_name_mapping_collision_should_duplicate_value() { + String response = + "{\n" + + " \"composite#composite_buckets\": {\n" + + " \"buckets\": [\n" + + " {\n" + + " \"key\": { \"dept_id\": \"eng\" },\n" + + " \"doc_count\": 3,\n" + + " \"top_hits#dedup\": {\n" + + " \"hits\": {\n" + + " \"total\": { \"value\": 3, \"relation\": \"eq\" },\n" + + " \"hits\": [\n" + + " {\n" + + " \"_index\": \"idx\",\n" + + " \"_id\": \"1\",\n" + + " \"fields\": { \"salary\": [50000] }\n" + + " }\n" + + " ]\n" + + " }\n" + + " }\n" + + " }\n" + + " ]\n" + + " }\n" + + "}"; + + // salary -> [pay, pay2]: both rename and eval-column-ref resolve to the same source field + Map> mapping = Map.of("salary", List.of("pay", "pay2")); + OpenSearchAggregationResponseParser parser = + new CompositeAggregationParser(new TopHitsParser("dedup", false, false, mapping)); + + List> result = parse(parser, response); + // Bucket key row + assertThat(result.get(0), org.hamcrest.Matchers.hasEntry("dept_id", "eng")); + // Hit row must contain both aliases with the same value; original key must be absent + Map hitRow = result.get(1); + assertEquals(50000, hitRow.get("pay")); + assertEquals(50000, hitRow.get("pay2")); + assertNull(hitRow.get("salary"), "original field name must be removed after mapping"); + } + public List> parse(OpenSearchAggregationResponseParser parser, String json) { return parser.parse(fromJson(json)); }