Remove prefix-override branch in shouldOverrideField; scope struct-parent pruning to bin

PR #4606 introduced a `newName.startsWith(originalName + ".")` branch in
`shouldOverrideField` to make `bin` on a nested field work without an explicit
`fields` projection (issue #4482). That heuristic silently removed any column
that happened to be a prefix of an eval target, which broke two real cases:

* Issue #5185 — assigning multiple dotted-path fields from a `spath` MAP root
  dropped the root column, so subsequent evals failed with "Field not found".
* Reviewer's case on PR #5351 —
  `source=big5 | fields agent | eval agent.name = "test"` dropped the `agent`
  column entirely, violating SPL1 semantics where a dotted assignment creates
  a literal column without touching any sibling or parent field.

Rather than layer another heuristic (`hasSubFields`) on top of the prefix
branch, this change removes the prefix branch entirely and returns
`shouldOverrideField` to an exact-name match. The command-specific behaviour
that actually needed parent-pruning — `bin` on a nested leaf inside a
struct that OpenSearch exposes as both parent and flattened leaves — is now
handled locally in `visitBin` via `dropStructParentsFor`, which only fires
when the prefixes really exist as struct-parent columns in the current row
schema.

Tests:
* Kept the two spath regression tests from PR #5351
  (`CalcitePPLSpathTest`, `CalcitePPLSpathCommandIT`,
  `issues/5185.yml`).
* Added `CalciteEvalCommandIT.testEvalDottedNameDoesNotDropStructParent` and
  `testEvalDottedNamePreservesStructParent_ImplicitProject` to cover the
  reviewer's `agent` case.
* `CalciteBinCommandIT.testBinWithNestedFieldWithoutExplicitProjection` and
  `testBinWithEvalCreatedDottedFieldName` still pass via the new
  `dropStructParentsFor` path instead of the global prefix branch.

Resolves #5185

Signed-off-by: Songkan Tang <songkant@amazon.com>

Author: songkant-aws

core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java

@@ -973,10 +973,47 @@ public RelNode visitBin(Bin node, CalcitePlanContext context) {
 
     String alias = node.getAlias() != null ? node.getAlias() : fieldName;
     projectPlusOverriding(List.of(binExpression), List.of(alias), context);
+    dropStructParentsFor(alias, context);
 
     return context.relBuilder.peek();
   }
 
+  /**
+   * If {@code dottedName} addresses a nested leaf inside a struct that OpenSearch has exposed
+   * through both its struct-parent columns and its flattened leaf columns (e.g. the telemetry
+   * mapping exposes {@code resource}, {@code resource.attributes}, ..., {@code
+   * resource.attributes.telemetry.sdk.version} side-by-side), drop the struct-parent prefixes from
+   * the current row. This keeps a subsequent {@link #tryToRemoveNestedFields(CalcitePlanContext)}
+   * pass from collapsing the flattened leaves back into the parents when the final implicit {@code
+   * fields *} projection runs.
+   *
+   * <p>This preserves the behaviour that issue #4482 originally required for {@code bin} on a
+   * nested field without an explicit {@code fields} projection, scoped to the command that actually
+   * needs it instead of a global prefix-override in {@link #shouldOverrideField} (which broke #5185
+   * and the reviewer's {@code eval agent.name = ...} case).
+   *
+   * <p>No-op when no such struct-parent columns exist (e.g. flat columns or MAP roots from {@code
+   * spath}).
+   */
+  private void dropStructParentsFor(String dottedName, CalcitePlanContext context) {
+    if (dottedName == null || dottedName.indexOf('.') < 0) {
+      return;
+    }
+    List<String> fieldNames = context.relBuilder.peek().getRowType().getFieldNames();
+    List<RexNode> parentsToDrop = new ArrayList<>();
+    int dotIdx = dottedName.indexOf('.');
+    while (dotIdx >= 0) {
+      String prefix = dottedName.substring(0, dotIdx);
+      if (fieldNames.contains(prefix)) {
+        parentsToDrop.add(context.relBuilder.field(prefix));
+      }
+      dotIdx = dottedName.indexOf('.', dotIdx + 1);
+    }
+    if (!parentsToDrop.isEmpty()) {
+      context.relBuilder.projectExcept(parentsToDrop);
+    }
+  }
+
   @Override
   public RelNode visitParse(Parse node, CalcitePlanContext context) {
     visitChildren(node, context);
@@ -1295,7 +1332,7 @@ private void projectPlusOverriding(
     List<String> originalFieldNames = context.relBuilder.peek().getRowType().getFieldNames();
     List<RexNode> toOverrideList =
         originalFieldNames.stream()
-            .filter(originalName -> shouldOverrideField(originalName, newNames, originalFieldNames))
+            .filter(originalName -> shouldOverrideField(originalName, newNames))
             .map(a -> (RexNode) context.relBuilder.field(a))
             .toList();
     // 1. add the new fields, For example "age0, country0"
@@ -1315,31 +1352,30 @@ private void projectPlusOverriding(
     context.relBuilder.rename(expectedRenameFields);
   }
 
-  private boolean shouldOverrideField(
-      String originalName, List<String> newNames, List<String> allFieldNames) {
-    return newNames.stream()
-        .anyMatch(
-            newName ->
-                // Match exact field names (e.g., "age" == "age") for flat fields
-                newName.equals(originalName)
-                    // OR match nested paths (e.g., "resource.attributes..." starts with
-                    // "resource."), but only when the schema already contains sub-fields of
-                    // originalName. This distinguishes real struct parents (which have flattened
-                    // sub-field columns) from MAP columns produced by spath where dotted names
-                    // are just flat column names, not nested sub-fields.
-                    || (newName.startsWith(originalName + ".")
-                        && hasSubFields(originalName, allFieldNames)));
-  }
-
   /**
-   * Checks whether the schema contains any field whose name starts with {@code parentName + "."}.
-   * This indicates that {@code parentName} is a real struct parent with flattened sub-field
-   * columns, as opposed to a standalone MAP column (e.g., from spath) that has no sub-fields in the
-   * schema.
+   * Determine whether the column {@code originalName} should be replaced when a batch of new
+   * columns named {@code newNames} is being added. Only exact-name matches count as overrides —
+   * {@code eval foo.bar = ...} creates a brand new field literally named {@code foo.bar} and must
+   * never drop sibling or parent fields. This mirrors SPL1 semantics, where assigning a dotted name
+   * introduces a literal column of that name without touching any other field.
+   *
+   * <p>Earlier revisions (see PR #4606 / #5351) attempted to broaden this to a {@code
+   * newName.startsWith(originalName + ".")} prefix match. That prefix branch silently dropped any
+   * column that happened to be a prefix of an eval target, which caused two regressions:
+   *
+   * <ul>
+   *   <li>Issue #5185 — a MAP-typed root column produced by {@code spath} got dropped when eval
+   *       introduced multiple dotted-path fields under it.
+   *   <li>The reviewer's case on PR #5351 — {@code source=big5 | fields agent | eval agent.name =
+   *       "test"} dropped the {@code agent} column entirely.
+   * </ul>
+   *
+   * Command-specific behaviour that genuinely wants to prune struct-parent columns (such as {@code
+   * bin} on a nested field, the original motivation for #4606) must handle that inside the command
+   * visitor — see {@link #dropStructParentsFor(String, CalcitePlanContext)}.
    */
-  private boolean hasSubFields(String parentName, List<String> allFieldNames) {
-    String prefix = parentName + ".";
-    return allFieldNames.stream().anyMatch(name -> name.startsWith(prefix));
+  private boolean shouldOverrideField(String originalName, List<String> newNames) {
+    return newNames.contains(originalName);
   }
 
   private List<List<RexInputRef>> extractInputRefList(List<RelBuilder.AggCall> aggCalls) {

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

@@ -11,6 +11,7 @@
 import static org.opensearch.sql.util.MatcherUtils.verifyDataRows;
 import static org.opensearch.sql.util.MatcherUtils.verifySchema;
 
+import com.google.common.collect.ImmutableMap;
 import java.io.IOException;
 import org.json.JSONObject;
 import org.junit.jupiter.api.Test;
@@ -38,6 +39,21 @@ public void init() throws Exception {
     Request request3 = new Request("PUT", "/test_eval/_doc/3?refresh=true");
     request3.setJsonEntity("{\"name\": \"Charlie\", \"age\": null, \"title\": \"Analyst\"}");
     client().performRequest(request3);
+
+    // Index with a struct field `agent` to reproduce the reviewer's case from PR #5351:
+    //   source=<idx with agent struct> | fields agent | eval agent.name = "test"
+    // Rely on dynamic mapping — OpenSearch infers `agent` as an object with string children
+    // from the document contents. Using dynamic mapping keeps the init idempotent across
+    // repeated `@Before` invocations in the preserved cluster.
+    Request agentDoc1 = new Request("PUT", "/test_eval_agent/_doc/1?refresh=true");
+    agentDoc1.setJsonEntity(
+        "{\"agent\": {\"name\": \"winlogbeat\", \"version\": \"7.0\"}, \"message\": \"hello\"}");
+    client().performRequest(agentDoc1);
+
+    Request agentDoc2 = new Request("PUT", "/test_eval_agent/_doc/2?refresh=true");
+    agentDoc2.setJsonEntity(
+        "{\"agent\": {\"name\": \"filebeat\", \"version\": \"8.1\"}, \"message\": \"world\"}");
+    client().performRequest(agentDoc2);
   }
 
   @Test
@@ -86,6 +102,39 @@ public void testEvalStringConcatenationWithLiterals() throws IOException {
         rows("Charlie", "Analyst", "Name: Charlie, Title: Analyst"));
   }
 
+  @Test
+  public void testEvalDottedNameDoesNotDropStructParent() throws IOException {
+    // Reviewer's case from PR #5351: assigning a new dotted-path column must not remove the
+    // struct-parent column that happens to be a prefix of the eval target.
+    // Equivalent SPL1 query:
+    //   source=<idx with agent struct> | fields agent | eval agent.name = "test"
+    // Before the fix, the prefix-override in shouldOverrideField silently dropped the `agent`
+    // column entirely from the result schema. With the fix, `agent` is preserved.
+    // The newly-created literal column `agent.name` is also available (verified via an
+    // explicit trailing `fields` projection that bypasses tryToRemoveNestedFields).
+    JSONObject result =
+        executeQuery(
+            "source=test_eval_agent | fields agent | eval `agent.name` = 'test' | fields agent,"
+                + " `agent.name`");
+    verifySchema(result, schema("agent", "struct"), schema("agent.name", "string"));
+    verifyDataRows(
+        result,
+        rows(ImmutableMap.of("name", "winlogbeat", "version", "7.0"), "test"),
+        rows(ImmutableMap.of("name", "filebeat", "version", "8.1"), "test"));
+  }
+
+  @Test
+  public void testEvalDottedNamePreservesStructParent_ImplicitProject() throws IOException {
+    // Complementary coverage for the reviewer's case without the explicit trailing projection.
+    // With the implicit `fields *` (AllFields) that the PPL parser appends, the downstream
+    // `tryToRemoveNestedFields` pass still collapses the flattened leaf back into its struct
+    // parent -- but the important regression guard is that the struct parent `agent` is no
+    // longer dropped by `shouldOverrideField`'s prefix branch.
+    JSONObject result =
+        executeQuery("source=test_eval_agent | fields agent | eval `agent.name` = 'test'");
+    verifySchema(result, schema("agent", "struct"));
+  }
+
   @Test
   public void testEvalStringConcatenationWithExistingData() throws IOException {
     JSONObject result =