Remove redundant Jackson type polymorphism for DataSourceResult

The direct-query transport already carries the concrete result type as a
separate 'dataSourceType' string and dispatches on it via a switch in
both writeTo() and the StreamInput constructor. The Jackson polymorphism
on DataSourceResult (@JsonTypeInfo/@JsonSubTypes) was therefore never
exercised through the interface, and the earlier @JsonTypeInfo(Id.NONE)
workaround plus the addTypeFieldToJson()/contains("\"type\":") guard
existed only to satisfy that unused machinery -- which is what let a
Prometheus metric label named 'type' break deserialization (#5684).

Remove the polymorphism scaffolding entirely: drop the discriminator
annotations, read PrometheusResult directly, and delete the type
injection. Add a writeTo()->StreamInput round-trip test that carries a
metric label named 'type' to cover the transport path.

Note: this changes the serialized shape (no root 'type' in the result
JSON). Dispatch is unaffected (the separate dataSourceType string), and
the API is @opensearch.experimental, so the only impact is a bounded,
one-directional mixed-version window during a rolling upgrade.

Signed-off-by: Nagendra Mohan <nagendramohan1990@gmail.com>

Author: nagendramohan

direct-query/src/main/java/org/opensearch/sql/directquery/transport/model/ExecuteDirectQueryActionResponse.java

@@ -131,19 +131,11 @@ private Map<String, DataSourceResult> parseResult(
     Map<String, DataSourceResult> parsedResults = new HashMap<>();
 
     try {
-      // Add type to JSON if it doesn't already have it
-      final String resultWithType;
-      if (!rawResult.contains("\"type\":")) {
-        resultWithType = addTypeFieldToJson(rawResult, dataSourceType);
-      } else {
-        resultWithType = rawResult;
-      }
-
       DataSourceResult result;
       // Parse based on the determined data source type
       switch (dataSourceType.toLowerCase()) {
         case "prometheus":
-          result = OBJECT_MAPPER.readValue(resultWithType, PrometheusResult.class);
+          result = OBJECT_MAPPER.readValue(rawResult, PrometheusResult.class);
           break;
           // Add cases for other data source types as they're implemented
         default:
@@ -163,15 +155,4 @@ private Map<String, DataSourceResult> parseResult(
 
     return parsedResults;
   }
-
-  /**
-   * Adds a type field to the JSON string for proper polymorphic deserialization.
-   *
-   * @param rawJson The raw JSON string without a type field
-   * @param type The type to add
-   * @return Modified JSON string with type field
-   */
-  private String addTypeFieldToJson(String rawJson, String type) {
-    return rawJson.replaceFirst("\\{", "{\"type\":\"" + type + "\",");
-  }
 }

direct-query/src/main/java/org/opensearch/sql/directquery/transport/model/datasource/DataSourceResult.java

@@ -5,15 +5,14 @@
 
 package org.opensearch.sql.directquery.transport.model.datasource;
 
-import com.fasterxml.jackson.annotation.JsonSubTypes;
-import com.fasterxml.jackson.annotation.JsonTypeInfo;
-
 /**
  *
  * @opensearch.experimental
  *
  * Interface for results from various data sources.
+ *
+ * <p>Concrete result types are dispatched by the {@code dataSourceType} string carried alongside
+ * the serialized payload in the transport protocol (see {@code ExecuteDirectQueryActionResponse}),
+ * so no in-JSON Jackson type discriminator is used here.
  */
-@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "type")
-@JsonSubTypes({@JsonSubTypes.Type(value = PrometheusResult.class, name = "prometheus")})
 public interface DataSourceResult {}

direct-query/src/main/java/org/opensearch/sql/directquery/transport/model/datasource/PrometheusResult.java

@@ -7,8 +7,6 @@
 
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.annotation.JsonTypeInfo;
-import com.fasterxml.jackson.annotation.JsonTypeName;
 import java.util.List;
 import java.util.Map;
 import lombok.Getter;
@@ -22,8 +20,6 @@
  */
 @Getter
 @Setter
-@JsonTypeName("prometheus")
-@JsonTypeInfo(use = JsonTypeInfo.Id.NONE)
 @JsonIgnoreProperties(ignoreUnknown = true)
 public class PrometheusResult implements DataSourceResult {
 

direct-query/src/test/java/org/opensearch/sql/directquery/transport/model/ExecuteDirectQueryActionResponseTest.java

@@ -15,6 +15,7 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import org.junit.jupiter.api.Test;
 import org.opensearch.core.common.io.stream.InputStreamStreamInput;
@@ -147,6 +148,43 @@ public void testStreamSerializationWithPrometheusResult() throws IOException {
     assertInstanceOf(PrometheusResult.class, deserializedResponse.getResults().get("prom-ds-1"));
   }
 
+  @Test
+  public void testStreamSerializationPreservesTypeMetricLabel() throws IOException {
+    // Regression for #5684: a metric label literally named "type" must survive the
+    // writeTo() -> StreamInput round-trip (the transport path), not only parseResult().
+    Map<String, DataSourceResult> results = new HashMap<>();
+    PrometheusResult prometheusResult = new PrometheusResult();
+    prometheusResult.setResultType("vector");
+    PrometheusResult.PrometheusResultItem item = new PrometheusResult.PrometheusResultItem();
+    Map<String, String> metric = new HashMap<>();
+    metric.put("type", "counter");
+    metric.put("__name__", "http_requests_total");
+    item.setMetric(metric);
+    prometheusResult.setResult(List.of(item));
+    results.put("prom-ds-1", prometheusResult);
+
+    ExecuteDirectQueryActionResponse response =
+        new ExecuteDirectQueryActionResponse("query-type-label", results, "session-type-label");
+
+    ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+    StreamOutput streamOutput = new OutputStreamStreamOutput(outputStream);
+    response.writeTo(streamOutput);
+    streamOutput.close();
+
+    StreamInput streamInput =
+        new InputStreamStreamInput(new ByteArrayInputStream(outputStream.toByteArray()));
+    ExecuteDirectQueryActionResponse deserialized =
+        new ExecuteDirectQueryActionResponse(streamInput);
+    streamInput.close();
+
+    PrometheusResult out = (PrometheusResult) deserialized.getResults().get("prom-ds-1");
+    assertInstanceOf(PrometheusResult.class, out);
+    assertEquals("vector", out.getResultType());
+    assertEquals(1, out.getResult().size());
+    assertEquals("counter", out.getResult().get(0).getMetric().get("type"));
+    assertEquals("http_requests_total", out.getResult().get(0).getMetric().get("__name__"));
+  }
+
   @Test
   public void testStreamSerializationWithNullSessionId() throws IOException {
     String queryId = "query-no-session";