Fix dedup 500 on cross-index object/scalar mapping conflict (opensearch-project#5685)

On the Calcite path, `dedup` pushes down as a composite aggregation whose
surviving row is fetched via top_hits and decoded from the nested _source.
When a wildcard spans indices that disagree on whether a path is an object
or a scalar, the merged mapping types the path as an object (STRUCT) while
some documents store a scalar there. Decoding that scalar as a struct threw
`java.sql.SQLException: ... class java.lang.String cannot be cast to
class java.util.Map`, failing the whole query (100% on the reporting
customer's `source=logs-pr172502-*`).

Guard the STRUCT branch of OpenSearchExprValueFactory.parse to degrade a
scalar-under-object value to null, completing the graceful handling opensearch-project#5618
added for the geo_point and scalar branches. The catch is scoped to
ClassCastException because this branch is also the whole-document parse
entry point and each nested object level is wrapped independently, so a
broader catch would swallow legitimate errors raised deeper in the recursion.

Adds a factory unit regression plus two CalcitePPLDedupIT cases (object-vs-
scalar and the mirror scalar-vs-object, the latter guarding opensearch-project#5618).

Signed-off-by: Kai Huang <ahkcs@amazon.com>

Author: ahkcs

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLDedupIT.java

@@ -13,6 +13,8 @@
 import java.io.IOException;
 import org.json.JSONObject;
 import org.junit.jupiter.api.Test;
+import org.opensearch.client.Request;
+import org.opensearch.sql.legacy.TestUtils;
 import org.opensearch.sql.ppl.PPLIntegTestCase;
 import org.opensearch.sql.util.RequiresCapability;
 
@@ -27,6 +29,87 @@ public void init() throws Exception {
     loadIndex(Index.ACCOUNT);
   }
 
+  /**
+   * Regression test for https://github.com/opensearch-project/sql/issues/5685.
+   *
+   * <p>When {@code dedup} pushes down as a composite aggregation, the surviving row is fetched via
+   * {@code top_hits}, whose nested {@code _source} is parsed back into a struct. If a wildcard
+   * spans indices that disagree on whether an object path is an object or a scalar, the merged
+   * mapping types the path as an object while some documents store a scalar there — and parsing
+   * that scalar as a struct threw {@code java.sql.SQLException: ... String cannot be cast to ...
+   * Map}, failing the whole query (100% on Fidelity's {@code source=logs-pr172502-*}). The query
+   * must instead succeed and dedup on the requested key.
+   */
+  @Test
+  public void testDedupOverConflictingObjectScalarMapping() throws IOException {
+    // idx A (dynamic:false): only maps `host`; keeps resource.attributes.k8s.namespace as a scalar
+    //   in _source (unmapped). idx B maps that same path as an OBJECT with a keyword leaf `name`.
+    // Only B maps `resource`, so the wildcard merge classifies resource.attributes.k8s.namespace
+    // as STRUCT deterministically, while A's doc holds a scalar there.
+    if (!TestUtils.isIndexExist(client(), "dd5685_a")) {
+      TestUtils.createIndexByRestClient(
+          client(),
+          "dd5685_a",
+          "{\"mappings\":{\"dynamic\":false,\"properties\":{\"host\":{\"type\":\"keyword\"}}}}");
+      Request a1 = new Request("PUT", "/dd5685_a/_doc/1?refresh=true");
+      a1.setJsonEntity(
+          "{\"host\":\"h1\",\"resource\":{\"attributes\":{\"k8s\":{\"namespace\":\"prod\"}}}}");
+      client().performRequest(a1);
+    }
+    if (!TestUtils.isIndexExist(client(), "dd5685_b")) {
+      TestUtils.createIndexByRestClient(
+          client(),
+          "dd5685_b",
+          "{\"mappings\":{\"properties\":{\"host\":{\"type\":\"keyword\"},\"resource\":"
+              + "{\"properties\":{\"attributes\":{\"properties\":{\"k8s\":{\"properties\":"
+              + "{\"namespace\":{\"properties\":{\"name\":{\"type\":\"keyword\"}}}}}}}}}}}}");
+      Request b1 = new Request("PUT", "/dd5685_b/_doc/1?refresh=true");
+      b1.setJsonEntity(
+          "{\"host\":\"h2\",\"resource\":{\"attributes\":{\"k8s\":{\"namespace\":{\"name\":\"pod-x\"}}}}}");
+      client().performRequest(b1);
+    }
+    JSONObject actual = executeQuery("source=dd5685_* | dedup host | fields host");
+    verifyDataRows(actual, rows("h1"), rows("h2"));
+  }
+
+  /**
+   * Mirror of {@link #testDedupOverConflictingObjectScalarMapping}: the reverse conflict, where the
+   * merged mapping types a path as a scalar but some document stores an object there. Parsing that
+   * object under a scalar type must not throw {@code ... Map cannot be cast to ... String}.
+   * (Guarded upstream by the scalar-branch handling in #5618; kept here as an end-to-end
+   * regression.)
+   */
+  @Test
+  public void testDedupOverConflictingScalarObjectMapping() throws IOException {
+    // idx A maps resource.attributes.k8s.namespace as a scalar keyword; idx B (dynamic:false) does
+    //   not map it but stores an OBJECT there in _source. Only A maps the path, so the merged type
+    //   is keyword while B's doc holds an object under it.
+    if (!TestUtils.isIndexExist(client(), "dd5685c_a")) {
+      TestUtils.createIndexByRestClient(
+          client(),
+          "dd5685c_a",
+          "{\"mappings\":{\"properties\":{\"host\":{\"type\":\"keyword\"},\"resource\":"
+              + "{\"properties\":{\"attributes\":{\"properties\":{\"k8s\":{\"properties\":"
+              + "{\"namespace\":{\"type\":\"keyword\"}}}}}}}}}}");
+      Request a1 = new Request("PUT", "/dd5685c_a/_doc/1?refresh=true");
+      a1.setJsonEntity(
+          "{\"host\":\"h1\",\"resource\":{\"attributes\":{\"k8s\":{\"namespace\":\"prod\"}}}}");
+      client().performRequest(a1);
+    }
+    if (!TestUtils.isIndexExist(client(), "dd5685c_b")) {
+      TestUtils.createIndexByRestClient(
+          client(),
+          "dd5685c_b",
+          "{\"mappings\":{\"dynamic\":false,\"properties\":{\"host\":{\"type\":\"keyword\"}}}}");
+      Request b1 = new Request("PUT", "/dd5685c_b/_doc/1?refresh=true");
+      b1.setJsonEntity(
+          "{\"host\":\"h2\",\"resource\":{\"attributes\":{\"k8s\":{\"namespace\":{\"name\":\"pod-x\"}}}}}");
+      client().performRequest(b1);
+    }
+    JSONObject actual = executeQuery("source=dd5685c_* | dedup host | fields host");
+    verifyDataRows(actual, rows("h1"), rows("h2"));
+  }
+
   @Test
   public void testDedup() throws IOException {
     JSONObject actual =

opensearch/src/main/java/org/opensearch/sql/opensearch/data/value/OpenSearchExprValueFactory.java

@@ -222,7 +222,20 @@ private ExprValue parse(
       return parseArray(content, field, type, supportArrays);
     } else if (type.equals(OpenSearchDataType.of(OpenSearchDataType.MappingType.Object))
         || type == STRUCT) {
-      return parseStruct(content, field, supportArrays);
+      // A field mapped as an object can still hold a scalar value in a particular document — most
+      // often when a wildcard/alias spans indices that disagree on whether a path is an object or a
+      // scalar, so the merged mapping types it as an object (STRUCT) while some documents store a
+      // scalar there. Parsing that scalar as a struct throws ClassCastException (String cannot be
+      // cast to Map); return null for that field instead of failing the whole request, matching the
+      // graceful handling already applied to the geo_point and scalar branches. See #5685. The
+      // catch is scoped to ClassCastException (not Exception) because this branch is also the entry
+      // point for whole-document parsing, and each nested object level is wrapped independently — a
+      // broader catch would swallow legitimate errors raised deeper in the recursion.
+      try {
+        return parseStruct(content, field, supportArrays);
+      } catch (ClassCastException e) {
+        return ExprNullValue.of();
+      }
     } else if (typeActionMap.containsKey(type)) {
       if (content.isArray()) {
         return parseArray(content, field, type, supportArrays);

opensearch/src/test/java/org/opensearch/sql/opensearch/data/value/OpenSearchExprValueFactoryTest.java

@@ -728,6 +728,20 @@ public void constructStruct() {
         constructFromObject("structV", ImmutableMap.of("id", 1, "state", "WA")));
   }
 
+  /**
+   * A field mapped as an object (STRUCT) can still carry a scalar value in a particular document —
+   * e.g. when a wildcard/alias spans indices that disagree on whether the path is an object or a
+   * scalar, so the merged mapping types it as an object while a document (in this aggregation /
+   * {@code top_hits} decode path, an {@code ObjectContent}) stores a scalar. Parsing that scalar as
+   * a struct must not throw {@code String cannot be cast to Map}; it degrades to null, matching the
+   * graceful handling of the scalar/geo_point branches. Regression test for #5685.
+   */
+  @Test
+  public void constructStructWithScalarValueReturnsNull() {
+    assertEquals(nullValue(), constructFromObject("structV", "prod"));
+    assertEquals(nullValue(), constructFromObject("structV", 1));
+  }
+
   @Test
   public void constructIP() {
     final String ipString = "192.168.0.1";