Synthetic source support for flattened fields (#94842)

Here we add synthetic source support for fields whose type is flattened.
Note that flattened fields and synthetic source have the following limitations,
all arising from the fact that in synthetic source we just see key/value pairs
when reconstructing the original object and have no type information in mappings:

* flattened fields use sorted set doc values of keywords, which means two things: 
   first we do not allow duplicate values, second we treat all values as keywords
* reconstructing array of objects results in nested objects (no array)
* reconstructing arrays with just one element results in a single-value field since we
   have no way to distinguish single-valued from multi-values fields other then looking
   at the count of values

docs/changelog/94842.yaml

@@ -0,0 +1,5 @@
+pr: 94842
+summary: Flattened field synthetic support
+area: TSDB
+type: feature
+issues: []

docs/reference/mapping/fields/synthetic-source.asciidoc

@@ -44,6 +44,7 @@ types:
 ** <<date-nanos-synthetic-source,`date_nanos`>>
 ** <<dense-vector-synthetic-source,`dense_vector`>>
 ** <<numeric-synthetic-source,`double`>>
+** <<flattened-synthetic-source, `flattened`>>
 ** <<numeric-synthetic-source,`float`>>
 ** <<geo-point-synthetic-source,`geo_point`>>
 ** <<numeric-synthetic-source,`half_float`>>

docs/reference/mapping/types/flattened.asciidoc

@@ -124,8 +124,8 @@ specify the <<mapping-store, `store`>> parameter in the mapping.
 [[search-fields-flattened]]
 ==== Retrieving flattened fields
 
-Field values and concrete subfields can be retrieved using the 
-<<search-fields-param,fields parameter>>. content. Since the `flattened` field maps an 
+Field values and concrete subfields can be retrieved using the
+<<search-fields-param,fields parameter>>. content. Since the `flattened` field maps an
 entire object with potentially many subfields as a single field, the response contains
 the unaltered structure from `_source`.
 
@@ -235,7 +235,7 @@ POST /my-index-000001/_bulk?refresh
 
 Because `labels` is a `flattened` field type, the entire object is mapped as a
 single field. To retrieve values from this sub-field in a Painless script, use
-the `doc['<field_name>.<sub-field_name>'].value` format. 
+the `doc['<field_name>.<sub-field_name>'].value` format.
 
 [source,painless]
 ----
@@ -307,3 +307,123 @@ The following mapping parameters are accepted:
     Whether <<full-text-queries,full text queries>> should split the input on
     whitespace when building a query for this field. Accepts `true` or `false`
     (default).
+
+[[flattened-synthetic-source]]
+==== Synthetic `_source`
+
+IMPORTANT: Synthetic `_source` is Generally Available only for TSDB indices
+(indices that have `index.mode` set to `time_series`). For other indices
+synthetic `_source` is in technical preview. Features in technical preview may
+be changed or removed in a future release. Elastic will apply best effort to fix
+any issues, but features in technical preview are not subject to the support SLA
+of official GA features.
+
+Flattened fields support <<synthetic-source,synthetic`_source`>> in their default
+configuration. Synthetic `_source` cannot be used with <<doc-values,`doc_values`>>
+disabled.
+
+Synthetic source always sorts alphabetically and de-duplicates flattened fields.
+For example:
+[source,console,id=synthetic-source-flattened-sorting-example]
+----
+PUT idx
+{
+  "mappings": {
+    "_source": { "mode": "synthetic" },
+    "properties": {
+      "flattened": { "type": "flattened" }
+    }
+  }
+}
+PUT idx/_doc/1
+{
+  "flattened": {
+    "field": [ "apple", "apple", "banana", "avocado", "10", "200", "AVOCADO", "Banana", "Tangerine" ]
+  }
+}
+----
+// TEST[s/$/\nGET idx\/_doc\/1?filter_path=_source\n/]
+
+Will become:
+[source,console-result]
+----
+{
+  "flattened": {
+    "field": [ "10", "200", "AVOCADO", "Banana", "Tangerine", "apple", "avocado", "banana" ]
+  }
+}
+----
+// TEST[s/^/{"_source":/ s/\n$/}/]
+
+Synthetic source always uses nested objects instead of array of objects.
+For example:
+[source,console,id=synthetic-source-flattened-array-example]
+----
+PUT idx
+{
+  "mappings": {
+    "_source": { "mode": "synthetic" },
+    "properties": {
+      "flattened": { "type": "flattened" }
+    }
+  }
+}
+PUT idx/_doc/1
+{
+  "flattened": {
+      "field": [
+        { "id": 1, "name": "foo" },
+        { "id": 2, "name": "bar" },
+        { "id": 3, "name": "baz" }
+      ]
+  }
+}
+----
+// TEST[s/$/\nGET idx\/_doc\/1?filter_path=_source\n/]
+
+Will become (note the nested objects instead of the "flattened" array):
+[source,console-result]
+----
+{
+    "flattened": {
+      "field": {
+          "id": [ "1", "2", "3" ],
+          "name": [ "bar", "baz", "foo" ]
+      }
+    }
+}
+----
+// TEST[s/^/{"_source":/ s/\n$/}/]
+
+Synthetic source always uses single-valued fields for one-element arrays.
+For example:
+[source,console,id=synthetic-source-flattened-single-value-example]
+----
+PUT idx
+{
+  "mappings": {
+    "_source": { "mode": "synthetic" },
+    "properties": {
+      "flattened": { "type": "flattened" }
+    }
+  }
+}
+PUT idx/_doc/1
+{
+  "flattened": {
+    "field": [ "foo" ]
+  }
+}
+----
+// TEST[s/$/\nGET idx\/_doc\/1?filter_path=_source\n/]
+
+Will become (note the nested objects instead of the "flattened" array):
+[source,console-result]
+----
+{
+  "flattened": {
+    "field": "foo"
+  }
+}
+----
+// TEST[s/^/{"_source":/ s/\n$/}/]

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/get/100_synthetic_source.yml

@@ -808,3 +808,165 @@ ip with ignore_malformed:
           - hot garbage # fields saved by ignore_malformed are sorted after doc values
           - 7
   - is_false: fields
+
+
+---
+flattened field:
+  - skip:
+      version: " - 8.7.99"
+      reason: support for synthetic source on flattened fields added in 8.8.0
+
+  - do:
+      indices.create:
+        index: test
+        body:
+          mappings:
+            _source:
+              mode: synthetic
+            properties:
+              flattened:
+                type: flattened
+              flattened_object_array:
+                type: flattened
+              empty_flattened:
+                type: flattened
+              single_flattened:
+                type: flattened
+
+  - do:
+      index:
+        index: test
+        id: 1
+        body:
+          flattened:
+            top1: "876"
+            dict1:
+              abc: def
+              opq: rst
+              hij: lmn
+              uvx: wyz
+              z:
+                first: one
+                second: two
+                third: [ 1, 2, 3 ]
+            list1: [ "789", "1011", "1213", "1213" ]
+          single_flattened:
+            field: value
+          flattened_object_array:
+            - 1:
+                id: 1
+                code: 1234
+            - 2:
+                id: 2
+                code: 1243
+
+  - do:
+      get:
+        index: test
+        id: 1
+
+  - match: { _index: "test" }
+  - match: { _id: "1" }
+  - match: { _version: 1 }
+  - match: { found: true }
+  - match:
+      _source:
+        flattened:
+          top1: "876"
+          dict1:
+            abc: def
+            opq: rst
+            hij: lmn
+            uvx: wyz
+            z:
+              first: one
+              second: two
+              # NOTE 1: synthetic source always returns de-duplicated keywords
+              third: [ "1", "2", "3" ]
+          # NOTE 2: the missing '1213' value below is not a mistake.
+          # Flattened fields use SortedSetDocValues, which means they discard duplicate key/value pairs
+          # like `flattened.list1` => `1213` which exists twice in the input document.
+          # NOTE 3: watch out for string vs numeric sorting
+          list1: [ "1011", "1213", "789" ]
+        single_flattened:
+          field: value
+        # NOTE 3: this field is returned as nasted objects instead of an array of objects. We have no
+        # way to distinguish if field is an array of objects or a set of nested objects.
+        flattened_object_array:
+          1:
+            id: "1"
+            code: "1234"
+          2:
+            id: "2"
+            code: "1243"
+
+  - is_false: fields
+
+---
+flattened field no doc values:
+  - skip:
+      version: " - 8.7.99"
+      reason: support for synthetic source on flattened fields added in 8.8.0
+
+  - do:
+      catch: /field \[flattened\] of type \[flattened\] doesn't support synthetic source because it doesn't have doc values/
+      indices.create:
+        index: test
+        body:
+          mappings:
+            _source:
+              mode: synthetic
+            properties:
+              flattened:
+                type: flattened
+                doc_values: false
+
+---
+flattened field with ignore_above:
+  - skip:
+      version: " - 8.7.99"
+      reason: support for synthetic source on flattened fields added in 8.8.0
+
+  - do:
+      indices.create:
+        index: test
+        body:
+          mappings:
+            _source:
+              mode: synthetic
+            properties:
+              field:
+                type: flattened
+                ignore_above: 10
+
+  - do:
+      index:
+        index: test
+        id: 1
+        body:
+          field:
+            key1:
+              key2: "key2"
+              key3: "key3_ignored"
+            key4: "key4_ignored"
+            key5:
+              key6: "key6_ignored"
+            key7: "key7"
+
+  - do:
+      get:
+        index: test
+        id: 1
+
+  - match: { _index: "test" }
+  - match: { _id: "1" }
+  - match: { _version: 1 }
+  - match: { found: true }
+  - match:
+      _source:
+        field:
+          key1:
+            key2: "key2"
+          key7: "key7"
+
+  - is_false: fields

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/tsdb/90_unsupported_operations.yml

@@ -233,46 +233,6 @@ aggregate on _id:
               terms:
                 field: _id
 
----
-synthetic source flattened field:
-  - skip:
-      version: " - 8.6.99"
-      reason: "synthetic source introduced in 8.7.0"
-
-  - do:
-      catch: /field \[k8s.pod.labels\] of type \[flattened\] doesn't support synthetic source/
-      indices.create:
-        index: test-flattened
-        body:
-          settings:
-            number_of_shards: 1
-            number_of_replicas: 0
-            index:
-              mode: time_series
-              routing_path: [ metricset, k8s.pod.uid ]
-              time_series:
-                start_time: 2021-04-28T00:00:00Z
-                end_time: 2021-04-29T00:00:00Z
-          mappings:
-            properties:
-              "@timestamp":
-                type: date
-              metricset:
-                type: keyword
-                time_series_dimension: true
-              k8s:
-                properties:
-                  pod:
-                    properties:
-                      uid:
-                        type: keyword
-                        time_series_dimension: true
-                      labels:
-                        type: flattened
-                      value:
-                        type: long
-                        time_series_metric: gauge
-
 ---
 synthetic source text field:
   - skip:

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

@@ -52,6 +52,7 @@ import org.elasticsearch.index.mapper.FieldMapper;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.Mapper;
 import org.elasticsearch.index.mapper.MapperBuilderContext;
+import org.elasticsearch.index.mapper.SourceLoader;
 import org.elasticsearch.index.mapper.SourceValueFetcher;
 import org.elasticsearch.index.mapper.StringFieldType;
 import org.elasticsearch.index.mapper.TextParams;
@@ -729,4 +730,18 @@ public final class FlattenedFieldMapper extends FieldMapper {
     public FieldMapper.Builder getMergeBuilder() {
         return new Builder(simpleName()).init(this);
     }
+
+    @Override
+    public SourceLoader.SyntheticFieldLoader syntheticFieldLoader() {
+        if (hasScript()) {
+            return SourceLoader.SyntheticFieldLoader.NOTHING;
+        }
+        if (fieldType().hasDocValues()) {
+            return new FlattenedSortedSetDocValuesSyntheticFieldLoader(name() + "._keyed", simpleName());
+        }
+
+        throw new IllegalArgumentException(
+            "field [" + name() + "] of type [" + typeName() + "] doesn't support synthetic source because it doesn't have doc values"
+        );
+    }
 }

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

@@ -0,0 +1,307 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.index.mapper.flattened;
+
+import org.apache.lucene.index.SortedSetDocValues;
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * A helper class that reconstructs the field including keys and values
+ * mapped by {@link FlattenedFieldMapper}. It looks at key/value pairs,
+ * parses them and uses the {@link XContentBuilder} to reconstruct the
+ * original object.
+ *
+ * Example: given the following set of sorted key value pairs read from
+ * doc values:
+ *
+ * "id": "AAAb12"
+ * "package.id": "102938"
+ * "root.asset_code": "abc"
+ * "root.from": "New York"
+ * "root.object.value": "10"
+ * "root.object.items": "102202929"
+ * "root.object.items": "290092911"
+ * "root.to": "Atlanta"
+ *
+ * it turns them into the corresponding object using {@link XContentBuilder}.
+ * If, for instance JSON is used, it generates the following after deserialization:
+ *
+ * `{
+ *     "id": "AAAb12",
+ *     "package": {
+ *         "id": "102938"
+ *     },
+ *     "root": {
+ *         "asset_code": "abc",
+ *         "from": "New York",
+ *         "object": {
+ *             "items": ["102202929", "290092911"],
+ *         },
+ *         "to": "Atlanta"
+ *     }
+ * }`
+ *
+ */
+class FlattenedFieldSyntheticWriterHelper {
+
+    private record Prefix(List<String> prefix) {
+
+        Prefix() {
+            this(Collections.emptyList());
+        }
+
+        Prefix(final String key) {
+            this(key.split("\\."));
+        }
+
+        Prefix(String[] keyAsArray) {
+            this(List.of(keyAsArray).subList(0, keyAsArray.length - 1));
+        }
+
+        private Prefix shared(final Prefix other) {
+            return shared(this.prefix, other.prefix);
+        }
+
+        private static Prefix shared(final List<String> curr, final List<String> next) {
+            final List<String> shared = new ArrayList<>();
+            for (int i = 0; i < Math.min(curr.size(), next.size()); i++) {
+                if (curr.get(i).equals(next.get(i))) {
+                    shared.add(curr.get(i));
+                }
+            }
+
+            return new Prefix(shared);
+        }
+
+        private Prefix diff(final Prefix other) {
+            return diff(this.prefix, other.prefix);
+        }
+
+        private static Prefix diff(final List<String> a, final List<String> b) {
+            if (a.size() > b.size()) {
+                return diff(b, a);
+            }
+            final List<String> diff = new ArrayList<>();
+            if (a.isEmpty()) {
+                diff.addAll(b);
+                return new Prefix(diff);
+            }
+            int i = 0;
+            for (; i < a.size(); i++) {
+                if (a.get(i).equals(b.get(i)) == false) {
+                    break;
+                }
+            }
+            for (; i < b.size(); i++) {
+                diff.add(b.get(i));
+            }
+            return new Prefix(diff);
+        }
+
+        private Prefix reverse() {
+            final Prefix p = new Prefix(this.prefix);
+            Collections.reverse(p.prefix);
+            return p;
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(this.prefix);
+        }
+
+        @Override
+        public boolean equals(Object obj) {
+            if (obj == null) {
+                return false;
+            }
+            if (getClass() != obj.getClass()) {
+                return false;
+            }
+            Prefix other = (Prefix) obj;
+            return Objects.equals(this.prefix, other.prefix);
+        }
+    }
+
+    private record Suffix(List<String> suffix) {
+
+        Suffix() {
+            this(Collections.emptyList());
+        }
+
+        Suffix(final String key) {
+            this(key.split("\\."));
+        }
+
+        Suffix(final String[] keyAsArray) {
+            this(List.of(keyAsArray).subList(keyAsArray.length - 1, keyAsArray.length));
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(this.suffix);
+        }
+
+        @Override
+        public boolean equals(Object obj) {
+            if (obj == null) {
+                return false;
+            }
+            if (getClass() != obj.getClass()) {
+                return false;
+            }
+            Suffix other = (Suffix) obj;
+            return Objects.equals(this.suffix, other.suffix);
+        }
+    }
+
+    private static class KeyValue {
+
+        public static final KeyValue EMPTY = new KeyValue(null, new Prefix(), new Suffix());
+        private final String value;
+        private final Prefix prefix;
+        private final Suffix suffix;
+
+        private KeyValue(final String value, final Prefix prefix, final Suffix suffix) {
+            this.value = value;
+            this.prefix = prefix;
+            this.suffix = suffix;
+        }
+
+        KeyValue(final BytesRef keyValue) {
+            this(FlattenedFieldParser.extractKey(keyValue).utf8ToString(), FlattenedFieldParser.extractValue(keyValue).utf8ToString());
+        }
+
+        private KeyValue(final String key, final String value) {
+            this(value, new Prefix(key), new Suffix(key));
+        }
+
+        public Prefix prefix() {
+            return this.prefix;
+        }
+
+        public Suffix suffix() {
+            return this.suffix;
+        }
+
+        public Prefix start(final KeyValue other) {
+            return this.prefix.diff(this.prefix.shared(other.prefix));
+        }
+
+        public Prefix end(final KeyValue other) {
+            return start(other).reverse();
+        }
+
+        public String value() {
+            assert this.value != null;
+            return this.value;
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(this.value, this.prefix, this.suffix);
+        }
+
+        @Override
+        public boolean equals(Object obj) {
+            if (obj == null) {
+                return false;
+            }
+            if (getClass() != obj.getClass()) {
+                return false;
+            }
+            KeyValue other = (KeyValue) obj;
+            return Objects.equals(this.value, other.value)
+                && Objects.equals(this.prefix, other.prefix)
+                && Objects.equals(this.suffix, other.suffix);
+        }
+    }
+
+    private final SortedSetDocValues dv;
+
+    FlattenedFieldSyntheticWriterHelper(final SortedSetDocValues dv) {
+        this.dv = dv;
+    }
+
+    void write(final XContentBuilder b) throws IOException {
+        KeyValue curr = new KeyValue(dv.lookupOrd(dv.nextOrd()));
+        KeyValue prev = KeyValue.EMPTY;
+        final List<String> values = new ArrayList<>();
+        values.add(curr.value());
+        for (int i = 1; i < dv.docValueCount(); i++) {
+            KeyValue next = new KeyValue(dv.lookupOrd(dv.nextOrd()));
+            writeObject(b, curr, next, curr.start(prev), curr.end(next), values);
+            values.add(next.value());
+            prev = curr;
+            curr = next;
+        }
+        if (values.isEmpty() == false) {
+            writeObject(b, curr, KeyValue.EMPTY, curr.start(prev), curr.end(KeyValue.EMPTY), values);
+        }
+    }
+
+    private void writeObject(
+        final XContentBuilder b,
+        final KeyValue currKeyValue,
+        final KeyValue nextKeyValue,
+        final Prefix startPrefix,
+        final Prefix endPrefix,
+        final List<String> values
+    ) throws IOException {
+        startObject(b, startPrefix.prefix);
+        if (currKeyValue.suffix.equals(nextKeyValue.suffix) == false) {
+            writeNestedObject(b, values, currKeyValue.suffix().suffix);
+        }
+        endObject(b, endPrefix.prefix);
+    }
+
+    private static void writeNestedObject(final XContentBuilder b, final List<String> values, final List<String> currSuffix)
+        throws IOException {
+        for (int i = 0; i < currSuffix.size() - 1; i++) {
+            b.startObject(currSuffix.get(i));
+        }
+        writeField(b, values, currSuffix);
+        for (int i = 0; i < currSuffix.size() - 1; i++) {
+            b.endObject();
+        }
+        values.clear();
+    }
+
+    private static void endObject(final XContentBuilder b, final List<String> objects) throws IOException {
+        for (final String ignored : objects) {
+            b.endObject();
+        }
+    }
+
+    private static void startObject(final XContentBuilder b, final List<String> objects) throws IOException {
+        for (final String object : objects) {
+            b.startObject(object);
+        }
+    }
+
+    private static void writeField(XContentBuilder b, List<String> values, List<String> currSuffix) throws IOException {
+        if (values.size() > 1) {
+            b.field(currSuffix.get(currSuffix.size() - 1), values);
+        } else {
+            // NOTE: here we make the assumption that fields with just one value are not arrays.
+            // Flattened fields have no mappings, and even if we had mappings, there is no way
+            // in Elasticsearch to distinguish single valued fields from multi-valued fields.
+            // As a result, there is no way to know, after reading a single value, if that value
+            // is the value for a single-valued field or a multi-valued field (array) with just
+            // one value (array of size 1).
+            b.field(currSuffix.get(currSuffix.size() - 1), values.get(0));
+        }
+    }
+}

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

@@ -0,0 +1,119 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.index.mapper.flattened;
+
+import org.apache.lucene.index.DocValues;
+import org.apache.lucene.index.LeafReader;
+import org.apache.lucene.index.SortedSetDocValues;
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.index.mapper.SortedSetDocValuesSyntheticFieldLoader;
+import org.elasticsearch.xcontent.XContentBuilder;
+
+import java.io.IOException;
+
+public class FlattenedSortedSetDocValuesSyntheticFieldLoader extends SortedSetDocValuesSyntheticFieldLoader {
+    private DocValuesFieldValues docValues = NO_VALUES;
+    private final String name;
+    private final String simpleName;
+
+    /**
+     * Build a loader for flattened fields from doc values.
+     *
+     * @param name                      the name of the field to load from doc values
+     * @param simpleName                the name to give the field in the rendered {@code _source}
+     */
+    public FlattenedSortedSetDocValuesSyntheticFieldLoader(String name, String simpleName) {
+        super(name, simpleName, null, false);
+        this.name = name;
+        this.simpleName = simpleName;
+    }
+
+    @Override
+    public DocValuesLoader docValuesLoader(LeafReader reader, int[] docIdsInLeaf) throws IOException {
+        final SortedSetDocValues dv = DocValues.getSortedSet(reader, name);
+        if (dv.getValueCount() == 0) {
+            docValues = NO_VALUES;
+            return null;
+        }
+        final FlattenedFieldDocValuesLoader loader = new FlattenedFieldDocValuesLoader(dv);
+        docValues = loader;
+        return loader;
+    }
+
+    @Override
+    public boolean hasValue() {
+        return docValues.count() > 0;
+    }
+
+    @Override
+    public void write(XContentBuilder b) throws IOException {
+        if (docValues.count() == 0) {
+            return;
+        }
+        b.startObject(simpleName);
+        docValues.write(b);
+        b.endObject();
+    }
+
+    @Override
+    protected BytesRef convert(BytesRef value) {
+        return value;
+    }
+
+    @Override
+    protected BytesRef preserve(BytesRef value) {
+        return BytesRef.deepCopyOf(value);
+    }
+
+    private interface DocValuesFieldValues {
+        int count();
+
+        void write(XContentBuilder b) throws IOException;
+    }
+
+    private static final DocValuesFieldValues NO_VALUES = new DocValuesFieldValues() {
+        @Override
+        public int count() {
+            return 0;
+        }
+
+        @Override
+        public void write(XContentBuilder b) {}
+    };
+
+    /**
+     * Load ordinals in line with populating the doc and immediately
+     * convert from ordinals into {@link BytesRef}s.
+     */
+    private static class FlattenedFieldDocValuesLoader implements DocValuesLoader, DocValuesFieldValues {
+        private final SortedSetDocValues dv;
+        private boolean hasValue;
+        private final FlattenedFieldSyntheticWriterHelper writer;
+
+        FlattenedFieldDocValuesLoader(final SortedSetDocValues dv) {
+            this.dv = dv;
+            this.writer = new FlattenedFieldSyntheticWriterHelper(dv);
+        }
+
+        @Override
+        public boolean advanceToDoc(int docId) throws IOException {
+            return hasValue = dv.advanceExact(docId);
+        }
+
+        @Override
+        public int count() {
+            return hasValue ? dv.docValueCount() : 0;
+        }
+
+        @Override
+        public void write(XContentBuilder b) throws IOException {
+            this.writer.write(b);
+        }
+    }
+}

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

@@ -30,9 +30,14 @@ import org.elasticsearch.xcontent.XContentType;
 import org.junit.AssumptionViolatedException;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
 
 import static org.apache.lucene.tests.analysis.BaseTokenStreamTestCase.assertTokenStreamContents;
 import static org.hamcrest.Matchers.containsString;
@@ -516,11 +521,68 @@ public class FlattenedFieldMapperTests extends MapperTestCase {
 
     @Override
     protected SyntheticSourceSupport syntheticSourceSupport(boolean ignoreMalformed) {
-        throw new AssumptionViolatedException("not supported");
+        return new FlattenedFieldSyntheticSourceSupport();
     }
 
     @Override
     protected IngestScriptSupport ingestScriptSupport() {
         throw new AssumptionViolatedException("not supported");
     }
+
+    private static void randomMapExample(final TreeMap<Object, Object> example, int depth, int maxDepth) {
+        for (int i = 0; i < randomIntBetween(2, 5); i++) {
+            int j = depth >= maxDepth ? randomIntBetween(1, 2) : randomIntBetween(1, 3);
+            switch (j) {
+                case 1 -> example.put(randomAlphaOfLength(10), randomAlphaOfLength(10));
+                case 2 -> {
+                    int size = randomIntBetween(2, 10);
+                    final Set<String> stringSet = new HashSet<>();
+                    while (stringSet.size() < size) {
+                        stringSet.add(String.valueOf(randomIntBetween(10_000, 20_000)));
+                    }
+                    final List<String> randomList = new ArrayList<>(stringSet);
+                    Collections.sort(randomList);
+                    example.put(randomAlphaOfLength(6), randomList);
+                }
+                case 3 -> {
+                    final TreeMap<Object, Object> nested = new TreeMap<>();
+                    randomMapExample(nested, depth + 1, maxDepth);
+                    example.put(randomAlphaOfLength(10), nested);
+                }
+                default -> throw new IllegalArgumentException("value: [" + j + "] unexpected");
+            }
+        }
+    }
+
+    private static class FlattenedFieldSyntheticSourceSupport implements SyntheticSourceSupport {
+
+        @Override
+        public SyntheticSourceExample example(int maxValues) throws IOException {
+
+            // NOTE: values must be keywords and we use a TreeMap to preserve order (doc values are sorted and the result
+            // is created with keys and nested keys in sorted order).
+            final TreeMap<Object, Object> map = new TreeMap<>();
+            randomMapExample(map, 0, maxValues);
+            return new SyntheticSourceExample(map, map, this::mapping);
+        }
+
+        @Override
+        public List<SyntheticSourceInvalidExample> invalidExample() throws IOException {
+            return List.of(
+                new SyntheticSourceInvalidExample(
+                    equalTo("field [field] of type [flattened] doesn't support synthetic " + "source because it doesn't have doc values"),
+                    b -> b.field("type", "flattened").field("doc_values", false)
+                )
+            );
+        }
+
+        private void mapping(XContentBuilder b) throws IOException {
+            b.field("type", "flattened");
+        }
+    }
+
+    @Override
+    protected boolean supportsCopyTo() {
+        return false;
+    }
 }

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

@@ -0,0 +1,193 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.index.mapper.flattened;
+
+import org.apache.lucene.index.SortedSetDocValues;
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.XContentType;
+import org.mockito.ArgumentMatchers;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class FlattenedFieldSyntheticWriterHelperTests extends ESTestCase {
+
+    public void testSingleField() throws IOException {
+        // GIVEN
+        final SortedSetDocValues dv = mock(SortedSetDocValues.class);
+        when(dv.getValueCount()).thenReturn(1L);
+        when(dv.docValueCount()).thenReturn(1);
+        byte[] bytes = ("test" + '\0' + "one").getBytes(StandardCharsets.UTF_8);
+        when(dv.nextOrd()).thenReturn(0L);
+        when(dv.lookupOrd(0L)).thenReturn(new BytesRef(bytes, 0, bytes.length));
+        FlattenedFieldSyntheticWriterHelper writer = new FlattenedFieldSyntheticWriterHelper(dv);
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        XContentBuilder b = new XContentBuilder(XContentType.JSON.xContent(), baos);
+
+        // WHEN
+        b.startObject();
+        writer.write(b);
+        b.endObject();
+        b.flush();
+
+        // THEN
+        assertEquals("{\"test\":\"one\"}", baos.toString(StandardCharsets.UTF_8));
+    }
+
+    public void testFlatObject() throws IOException {
+        // GIVEN
+        final SortedSetDocValues dv = mock(SortedSetDocValues.class);
+        final FlattenedFieldSyntheticWriterHelper writer = new FlattenedFieldSyntheticWriterHelper(dv);
+        final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        final XContentBuilder builder = new XContentBuilder(XContentType.JSON.xContent(), baos);
+        final List<byte[]> bytes = List.of("a" + '\0' + "value_a", "b" + '\0' + "value_b", "c" + '\0' + "value_c", "d" + '\0' + "value_d")
+            .stream()
+            .map(x -> x.getBytes(StandardCharsets.UTF_8))
+            .collect(Collectors.toList());
+        when(dv.getValueCount()).thenReturn(Long.valueOf(bytes.size()));
+        when(dv.docValueCount()).thenReturn(bytes.size());
+        when(dv.nextOrd()).thenReturn(0L, 1L, 2L, 3L);
+        for (int i = 0; i < bytes.size(); i++) {
+            when(dv.lookupOrd(ArgumentMatchers.eq((long) i))).thenReturn(new BytesRef(bytes.get(i), 0, bytes.get(i).length));
+        }
+
+        // WHEN
+        builder.startObject();
+        writer.write(builder);
+        builder.endObject();
+        builder.flush();
+
+        // THEN
+        assertEquals("{\"a\":\"value_a\",\"b\":\"value_b\",\"c\":\"value_c\",\"d\":\"value_d\"}", baos.toString(StandardCharsets.UTF_8));
+    }
+
+    public void testSingleObject() throws IOException {
+        // GIVEN
+        final SortedSetDocValues dv = mock(SortedSetDocValues.class);
+        final FlattenedFieldSyntheticWriterHelper writer = new FlattenedFieldSyntheticWriterHelper(dv);
+        final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        final XContentBuilder builder = new XContentBuilder(XContentType.JSON.xContent(), baos);
+        final List<byte[]> bytes = List.of(
+            "a" + '\0' + "value_a",
+            "a.b" + '\0' + "value_b",
+            "a.b.c" + '\0' + "value_c",
+            "a.d" + '\0' + "value_d"
+        ).stream().map(x -> x.getBytes(StandardCharsets.UTF_8)).collect(Collectors.toList());
+        when(dv.getValueCount()).thenReturn(Long.valueOf(bytes.size()));
+        when(dv.docValueCount()).thenReturn(bytes.size());
+        when(dv.nextOrd()).thenReturn(0L, 1L, 2L, 3L);
+        for (int i = 0; i < bytes.size(); i++) {
+            when(dv.lookupOrd(ArgumentMatchers.eq((long) i))).thenReturn(new BytesRef(bytes.get(i), 0, bytes.get(i).length));
+        }
+
+        // WHEN
+        builder.startObject();
+        writer.write(builder);
+        builder.endObject();
+        builder.flush();
+
+        // THEN
+        assertEquals(
+            "{\"a\":\"value_a\",\"a\":{\"b\":\"value_b\",\"b\":{\"c\":\"value_c\"},\"d\":\"value_d\"}}",
+            baos.toString(StandardCharsets.UTF_8)
+        );
+    }
+
+    public void testMultipleObjects() throws IOException {
+        // GIVEN
+        final SortedSetDocValues dv = mock(SortedSetDocValues.class);
+        final FlattenedFieldSyntheticWriterHelper writer = new FlattenedFieldSyntheticWriterHelper(dv);
+        final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        final XContentBuilder builder = new XContentBuilder(XContentType.JSON.xContent(), baos);
+        final List<byte[]> bytes = List.of("a.x" + '\0' + "10", "a.y" + '\0' + "20", "b.a" + '\0' + "30", "b.c" + '\0' + "40")
+            .stream()
+            .map(x -> x.getBytes(StandardCharsets.UTF_8))
+            .collect(Collectors.toList());
+        when(dv.getValueCount()).thenReturn(Long.valueOf(bytes.size()));
+        when(dv.docValueCount()).thenReturn(bytes.size());
+        when(dv.nextOrd()).thenReturn(0L, 1L, 2L, 3L);
+        for (int i = 0; i < bytes.size(); i++) {
+            when(dv.lookupOrd(ArgumentMatchers.eq((long) i))).thenReturn(new BytesRef(bytes.get(i), 0, bytes.get(i).length));
+        }
+
+        // WHEN
+        builder.startObject();
+        writer.write(builder);
+        builder.endObject();
+        builder.flush();
+
+        // THEN
+        assertEquals("{\"a\":{\"x\":\"10\",\"y\":\"20\"},\"b\":{\"a\":\"30\",\"c\":\"40\"}}", baos.toString(StandardCharsets.UTF_8));
+    }
+
+    public void testSingleArray() throws IOException {
+        // GIVEN
+        final SortedSetDocValues dv = mock(SortedSetDocValues.class);
+        final FlattenedFieldSyntheticWriterHelper writer = new FlattenedFieldSyntheticWriterHelper(dv);
+        final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        final XContentBuilder builder = new XContentBuilder(XContentType.JSON.xContent(), baos);
+        final List<byte[]> bytes = List.of("a.x" + '\0' + "10", "a.x" + '\0' + "20", "a.x" + '\0' + "30", "a.x" + '\0' + "40")
+            .stream()
+            .map(x -> x.getBytes(StandardCharsets.UTF_8))
+            .collect(Collectors.toList());
+        when(dv.getValueCount()).thenReturn(Long.valueOf(bytes.size()));
+        when(dv.docValueCount()).thenReturn(bytes.size());
+        when(dv.nextOrd()).thenReturn(0L, 1L, 2L, 3L);
+        for (int i = 0; i < bytes.size(); i++) {
+            when(dv.lookupOrd(ArgumentMatchers.eq((long) i))).thenReturn(new BytesRef(bytes.get(i), 0, bytes.get(i).length));
+        }
+
+        // WHEN
+        builder.startObject();
+        writer.write(builder);
+        builder.endObject();
+        builder.flush();
+
+        // THEN
+        assertEquals("{\"a\":{\"x\":[\"10\",\"20\",\"30\",\"40\"]}}", baos.toString(StandardCharsets.UTF_8));
+    }
+
+    public void testMultipleArrays() throws IOException {
+        // GIVEN
+        final SortedSetDocValues dv = mock(SortedSetDocValues.class);
+        final FlattenedFieldSyntheticWriterHelper writer = new FlattenedFieldSyntheticWriterHelper(dv);
+        final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        final XContentBuilder builder = new XContentBuilder(XContentType.JSON.xContent(), baos);
+        final List<byte[]> bytes = List.of(
+            "a.x" + '\0' + "10",
+            "a.x" + '\0' + "20",
+            "b.y" + '\0' + "30",
+            "b.y" + '\0' + "40",
+            "b.y" + '\0' + "50"
+        ).stream().map(x -> x.getBytes(StandardCharsets.UTF_8)).collect(Collectors.toList());
+        when(dv.getValueCount()).thenReturn(Long.valueOf(bytes.size()));
+        when(dv.docValueCount()).thenReturn(bytes.size());
+        when(dv.nextOrd()).thenReturn(0L, 1L, 2L, 3L, 4L);
+        for (int i = 0; i < bytes.size(); i++) {
+            when(dv.lookupOrd(ArgumentMatchers.eq((long) i))).thenReturn(new BytesRef(bytes.get(i), 0, bytes.get(i).length));
+        }
+
+        // WHEN
+        builder.startObject();
+        writer.write(builder);
+        builder.endObject();
+        builder.flush();
+
+        // THEN
+        assertEquals("{\"a\":{\"x\":[\"10\",\"20\"]},\"b\":{\"y\":[\"30\",\"40\",\"50\"]}}", baos.toString(StandardCharsets.UTF_8));
+    }
+}