Fix bin nested fields issue

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

Author: ahkcs

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

@@ -850,7 +850,16 @@ private void projectPlusOverriding(
     List<String> originalFieldNames = context.relBuilder.peek().getRowType().getFieldNames();
     List<RexNode> toOverrideList =
         originalFieldNames.stream()
-            .filter(newNames::contains)
+            .filter(
+                originalName ->
+                    newNames.stream()
+                        .anyMatch(
+                            newName ->
+                                // Match exact field names (e.g., "age" == "age")
+                                // OR nested paths (e.g., "resource.attributes..." starts with
+                                // "resource")
+                                newName.equals(originalName)
+                                    || newName.startsWith(originalName + ".")))
             .map(a -> (RexNode) context.relBuilder.field(a))
             .toList();
     // 1. add the new fields, For example "age0, country0"

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

@@ -27,6 +27,7 @@ public void init() throws Exception {
     loadIndex(Index.BANK);
     loadIndex(Index.EVENTS_NULL);
     loadIndex(Index.TIME_TEST_DATA);
+    loadIndex(Index.TELEMETRY);
   }
 
   @Test
@@ -984,4 +985,56 @@ public void testStatsWithBinsOnTimeAndTermField_Avg() throws IOException {
         rows(50, "us-east", "2024-07-01 00:05:00"),
         rows(40.25, "us-west", "2024-07-01 00:01:00"));
   }
+
+  @Test
+  public void testBinWithNestedFieldWithoutExplicitProjection() throws IOException {
+    // Test bin command on nested field without explicit fields projection
+    // This reproduces the bug from https://github.com/opensearch-project/sql/issues/4482
+    // The telemetry index has: resource.attributes.telemetry.sdk.version (values: 10, 11, 12, 13,
+    // 14)
+    JSONObject result =
+        executeQuery(
+            String.format(
+                "source=%s | bin `resource.attributes.telemetry.sdk.version` span=2 | sort"
+                    + " `resource.attributes.telemetry.sdk.version`",
+                TEST_INDEX_TELEMETRY));
+
+    // When binning a nested field, all sibling fields in the struct are also returned
+    verifySchema(
+        result,
+        schema("resource.attributes.telemetry.sdk.enabled", null, "boolean"),
+        schema("resource.attributes.telemetry.sdk.language", null, "string"),
+        schema("resource.attributes.telemetry.sdk.name", null, "string"),
+        schema("severityNumber", null, "int"),
+        schema("resource.attributes.telemetry.sdk.version", null, "string"));
+
+    // With span=2 on values [10, 11, 12, 13, 14], we expect binned ranges:
+    // 10 -> 10-12, 11 -> 10-12, 12 -> 12-14, 13 -> 12-14, 14 -> 14-16
+    // The binned field is the last column
+    verifyDataRows(
+        result,
+        rows(true, "java", "opentelemetry", 9, "10-12"),
+        rows(false, "python", "opentelemetry", 12, "10-12"),
+        rows(true, "javascript", "opentelemetry", 9, "12-14"),
+        rows(false, "go", "opentelemetry", 16, "12-14"),
+        rows(true, "rust", "opentelemetry", 12, "14-16"));
+  }
+
+  @Test
+  public void testBinWithNestedFieldWithExplicitProjection() throws IOException {
+    // Test bin command on nested field WITH explicit fields projection (workaround)
+    // This is the workaround mentioned in https://github.com/opensearch-project/sql/issues/4482
+    JSONObject result =
+        executeQuery(
+            String.format(
+                "source=%s | bin `resource.attributes.telemetry.sdk.version` span=2 | fields"
+                    + " `resource.attributes.telemetry.sdk.version` | sort"
+                    + " `resource.attributes.telemetry.sdk.version`",
+                TEST_INDEX_TELEMETRY));
+    verifySchema(result, schema("resource.attributes.telemetry.sdk.version", null, "string"));
+
+    // With span=2 on values [10, 11, 12, 13, 14], we expect binned ranges
+    verifyDataRows(
+        result, rows("10-12"), rows("10-12"), rows("12-14"), rows("12-14"), rows("14-16"));
+  }
 }