[9.0] Check prefixes when constructing synthetic source for flattened fields (#129580)

docs/changelog/129580.yaml

@@ -0,0 +1,6 @@
+pr: 129580
+summary: Check prefixes when constructing synthetic source for flattened fields
+area: Mapping
+type: bug
+issues:
+ - 129508

server/src/main/java/org/elasticsearch/index/mapper/flattened/FlattenedFieldSyntheticWriterHelper.java

@@ -261,7 +261,7 @@ private static void writeObject(
         final List<String> values
     ) throws IOException {
         startObject(b, startPrefix.prefix);
-        if (currKeyValue.suffix.equals(nextKeyValue.suffix) == false) {
+        if (currKeyValue.prefix.equals(nextKeyValue.prefix) == false || currKeyValue.suffix.equals(nextKeyValue.suffix) == false) {
             writeNestedObject(b, values, currKeyValue.suffix().suffix);
         }
         endObject(b, endPrefix.prefix);

server/src/test/java/org/elasticsearch/index/mapper/flattened/FlattenedFieldMapperTests.java

@@ -851,6 +851,23 @@ public void testSyntheticSourceWithOnlyIgnoredValues() throws IOException {
         assertThat(syntheticSource, equalTo("{\"field\":{\"key1\":\"val1\",\"obj1\":{\"key2\":\"val2\",\"key3\":[\"val3\",\"val4\"]}}}"));
     }
 
+    public void testSyntheticSourceWithCommonLeafField() throws IOException {
+        DocumentMapper mapper = createSytheticSourceMapperService(
+            mapping(b -> { b.startObject("field").field("type", "flattened").endObject(); })
+        ).documentMapper();
+
+        var syntheticSource = syntheticSource(mapper, b -> {
+            b.startObject("field");
+            {
+                b.startObject("obj1").field("key", "foo").endObject();
+                b.startObject("obj2").field("key", "bar").endObject();
+            }
+            b.endObject();
+        });
+        assertThat(syntheticSource, equalTo("""
+            {"field":{"obj1":{"key":"foo"},"obj2":{"key":"bar"}}}"""));
+    }
+
     @Override
     protected boolean supportsCopyTo() {
         return false;