fix(dedup): use Map<String,List<String>> for fieldNameMapping to handle alias collision

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
+   *
+   * <p>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
+   *
+   * <p>When both a {@code rename} and an {@code eval} column-reference resolve to the same original
+   * index field, the old {@code Map&lt;String,String&gt;} mapping silently dropped one alias on
+   * collision. With {@code Map&lt;String,List&lt;String&gt;&gt;} 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 {

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<String, List<String>> fieldNameMapping = new HashMap<>();
+        for (Pair<RexNode, String> 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(

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<String, List<String>> fieldNameMapping;
+
   public TopHitsParser(String name, boolean returnSingleValue, boolean returnMergeValue) {
+    this(name, returnSingleValue, returnMergeValue, null);
+  }
+
+  public TopHitsParser(
+      String name,
+      boolean returnSingleValue,
+      boolean returnMergeValue,
+      @Nullable Map<String, List<String>> fieldNameMapping) {
     this.name = name;
     this.returnSingleValue = returnSingleValue;
     this.returnMergeValue = returnMergeValue;
+    this.fieldNameMapping = fieldNameMapping;
   }
 
   @Override
@@ -129,12 +149,43 @@ public List<Map<String, Object>> 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.
+   *
+   * <p>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<String, Object> map) {
+    if (fieldNameMapping == null || fieldNameMapping.isEmpty()) {
+      return;
+    }
+    for (Map.Entry<String, List<String>> entry : fieldNameMapping.entrySet()) {
+      String originalName = entry.getKey();
+      Object value = map.get(originalName);
+      if (value == null && !map.containsKey(originalName)) {
+        continue;
+      }
+      List<String> 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);
   }

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}.
+   *
+   * <p>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<String, List<String>> 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<String,String>} approach silently dropped one mapping on collision; with {@code
+   * Map<String,List<String>>} both aliases must appear in the result.
+   *
+   * <p>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<String, List<String>> mapping = Map.of("salary", List.of("pay", "pay2"));
+    OpenSearchAggregationResponseParser parser =
+        new CompositeAggregationParser(new TopHitsParser("dedup", false, false, mapping));
+
+    List<Map<String, Object>> 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<String, Object> 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<Map<String, Object>> parse(OpenSearchAggregationResponseParser parser, String json) {
     return parser.parse(fromJson(json));
   }