EQL: Adds "fields" request field to the eql request (#68962)

client/rest-high-level/src/main/java/org/elasticsearch/client/eql/EqlSearchRequest.java

@@ -15,9 +15,11 @@ import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.search.fetch.subphase.FieldAndFormat;
 
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.List;
 import java.util.Objects;
 
 public class EqlSearchRequest implements Validatable, ToXContentObject {
@@ -29,6 +31,7 @@ public class EqlSearchRequest implements Validatable, ToXContentObject {
     private String timestampField = "@timestamp";
     private String eventCategoryField = "event.category";
     private String resultPosition = "tail";
+    private List<FieldAndFormat> fetchFields;
 
     private int size = 10;
     private int fetchSize = 1000;
@@ -51,6 +54,7 @@ public class EqlSearchRequest implements Validatable, ToXContentObject {
     static final String KEY_WAIT_FOR_COMPLETION_TIMEOUT = "wait_for_completion_timeout";
     static final String KEY_KEEP_ALIVE = "keep_alive";
     static final String KEY_KEEP_ON_COMPLETION = "keep_on_completion";
+    static final String KEY_FETCH_FIELDS = "fields";
 
     public EqlSearchRequest(String indices, String query) {
         indices(indices);
@@ -80,6 +84,9 @@ public class EqlSearchRequest implements Validatable, ToXContentObject {
             builder.field(KEY_KEEP_ALIVE, keepAlive);
         }
         builder.field(KEY_KEEP_ON_COMPLETION, keepOnCompletion);
+        if (fetchFields != null) {
+            builder.field(KEY_FETCH_FIELDS, fetchFields);
+        }
         builder.endObject();
         return builder;
     }
@@ -145,6 +152,15 @@ public class EqlSearchRequest implements Validatable, ToXContentObject {
         return this;
     }
 
+    public List<FieldAndFormat> fetchFields() {
+        return fetchFields;
+    }
+
+    public EqlSearchRequest fetchFields(List<FieldAndFormat> fetchFields) {
+        this.fetchFields = fetchFields;
+        return this;
+    }
+
     public int size() {
         return this.size;
     }
@@ -226,7 +242,8 @@ public class EqlSearchRequest implements Validatable, ToXContentObject {
             Objects.equals(waitForCompletionTimeout, that.waitForCompletionTimeout) &&
             Objects.equals(keepAlive, that.keepAlive) &&
             Objects.equals(keepOnCompletion, that.keepOnCompletion) &&
-            Objects.equals(resultPosition, that.resultPosition);
+            Objects.equals(resultPosition, that.resultPosition) &&
+            Objects.equals(fetchFields, that.fetchFields);
     }
 
     @Override
@@ -244,7 +261,8 @@ public class EqlSearchRequest implements Validatable, ToXContentObject {
             waitForCompletionTimeout,
             keepAlive,
             keepOnCompletion,
-            resultPosition);
+            resultPosition,
+            fetchFields);
     }
 
     public String[] indices() {

client/rest-high-level/src/test/java/org/elasticsearch/client/eql/EqlSearchResponseTests.java

@@ -11,17 +11,23 @@ package org.elasticsearch.client.eql;
 import org.apache.lucene.search.TotalHits;
 import org.elasticsearch.client.AbstractResponseTestCase;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.document.DocumentField;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.search.lookup.SourceLookup;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.RandomObjects;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.function.Supplier;
 
@@ -83,7 +89,15 @@ public class EqlSearchResponseTests extends AbstractResponseTestCase<org.elastic
             hits = new ArrayList<>();
             for (int i = 0; i < size; i++) {
                 BytesReference bytes = new RandomSource(() -> randomAlphaOfLength(10)).toBytes(xType);
-                hits.add(new org.elasticsearch.xpack.eql.action.EqlSearchResponse.Event(String.valueOf(i), randomAlphaOfLength(10), bytes));
+                Map<String, DocumentField> fetchFields = new HashMap<>();
+                for (int j = 0; j < randomIntBetween(0, 5); j++) {
+                    fetchFields.put(randomAlphaOfLength(10), randomDocumentField(xType).v1());
+                }
+                if (fetchFields.isEmpty() && randomBoolean()) {
+                    fetchFields = null;
+                }
+                hits.add(new org.elasticsearch.xpack.eql.action.EqlSearchResponse.Event(String.valueOf(i), randomAlphaOfLength(10), bytes,
+                    fetchFields));
             }
         }
         if (randomBoolean()) {
@@ -92,6 +106,30 @@ public class EqlSearchResponseTests extends AbstractResponseTestCase<org.elastic
         return hits;
     }
 
+    private static Tuple<DocumentField, DocumentField> randomDocumentField(XContentType xType) {
+        switch (randomIntBetween(0, 2)) {
+            case 0:
+                String fieldName = randomAlphaOfLengthBetween(3, 10);
+                Tuple<List<Object>, List<Object>> tuple = RandomObjects.randomStoredFieldValues(random(), xType);
+                DocumentField input = new DocumentField(fieldName, tuple.v1());
+                DocumentField expected = new DocumentField(fieldName, tuple.v2());
+                return Tuple.tuple(input, expected);
+            case 1:
+                List<Object> listValues = randomList(1, 5, () -> randomList(1, 5, ESTestCase::randomInt));
+                DocumentField listField = new DocumentField(randomAlphaOfLength(5), listValues);
+                return Tuple.tuple(listField, listField);
+            case 2:
+                List<Object> objectValues = randomList(1, 5, () ->
+                    Map.of(randomAlphaOfLength(5), randomInt(),
+                        randomAlphaOfLength(5), randomBoolean(),
+                        randomAlphaOfLength(5), randomAlphaOfLength(10)));
+                DocumentField objectField = new DocumentField(randomAlphaOfLength(5), objectValues);
+                return Tuple.tuple(objectField, objectField);
+            default:
+                throw new IllegalStateException();
+        }
+    }
+
     public static org.elasticsearch.xpack.eql.action.EqlSearchResponse createRandomEventsResponse(TotalHits totalHits, XContentType xType) {
         org.elasticsearch.xpack.eql.action.EqlSearchResponse.Hits hits = null;
         if (randomBoolean()) {

x-pack/plugin/eql/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/eql/10_basic.yml

@@ -1,5 +1,15 @@
 ---
 setup:
+  - do:
+      indices.create:
+          index:  eql_test
+          body:
+            mappings:
+              runtime:
+                day_of_week:
+                  type: keyword
+                  script:
+                    source: "emit(doc['@timestamp'].value.dayOfWeekEnum.getDisplayName(TextStyle.FULL, Locale.ROOT))"
   - do:
       bulk:
         refresh: true
@@ -49,6 +59,55 @@ setup:
   - match: {hits.events.1._id: "2"}
   - match: {hits.events.2._id: "3"}
 
+---
+"Execute EQL events query with fields filtering":
+  - do:
+      eql.search:
+        index: eql_test
+        body:
+          query: 'process where user == "SYSTEM"'
+          fields: [{"field":"@timestamp","format":"epoch_millis"},"id","valid","day_of_week"]
+
+  - match: {timed_out: false}
+  - match: {hits.total.value: 3}
+  - match: {hits.total.relation: "eq"}
+  - match: {hits.events.0._source.user: "SYSTEM"}
+  - match: {hits.events.0._id: "1"}
+  - match: {hits.events.0.fields.@timestamp: ["1580733296000"]}
+  - match: {hits.events.0.fields.id: [123]}
+  - match: {hits.events.0.fields.valid: [false]}
+  - match: {hits.events.0.fields.day_of_week: ["Monday"]}
+  - match: {hits.events.1._id: "2"}
+  - match: {hits.events.1.fields.@timestamp: ["1580819696000"]}
+  - match: {hits.events.1.fields.id: [123]}
+  - match: {hits.events.1.fields.valid: [true]}
+  - match: {hits.events.1.fields.day_of_week: ["Tuesday"]}
+  - match: {hits.events.2._id: "3"}
+  - match: {hits.events.2.fields.@timestamp: ["1580906096000"]}
+  - match: {hits.events.2.fields.id: [123]}
+  - match: {hits.events.2.fields.valid: [true]}
+  - match: {hits.events.2.fields.day_of_week: ["Wednesday"]}
+
+---
+"Execute EQL events query with filter_path":
+  - do:
+      eql.search:
+        index: eql_test
+        filter_path: "hits.events._source.event.category,hits.events.fields.user,hits.events.fields.id"
+        body:
+          query: 'process where user == "SYSTEM"'
+          fields: [{"field":"@timestamp","format":"epoch_millis"},"id","valid","user"]
+
+  - match: {hits.events.0._source.event.0.category: "process"}
+  - match: {hits.events.0.fields.id: [123]}
+  - match: {hits.events.0.fields.user: ["SYSTEM"]}
+  - match: {hits.events.1._source.event.0.category: "process"}
+  - match: {hits.events.1.fields.id: [123]}
+  - match: {hits.events.1.fields.user: ["SYSTEM"]}
+  - match: {hits.events.2._source.event.0.category: "process"}
+  - match: {hits.events.2.fields.id: [123]}
+  - match: {hits.events.2.fields.user: ["SYSTEM"]}
+
 ---
 "Execute EQL sequence with string key.":
   - do:
@@ -124,6 +183,57 @@ setup:
   - match: {hits.sequences.0.join_keys.0: true}
   - match: {hits.sequences.0.events.0._id: "2"}
   - match: {hits.sequences.0.events.1._id: "3"}
+
+---
+"Execute EQL sequence with fields filtering.":
+  - do:
+      eql.search:
+        index: eql_test
+        body:
+          query: 'sequence by user [process where user == "SYSTEM"] [process where true]'
+          fields: [{"field":"@timestamp","format":"epoch_millis"},"id","valid","day_of_week"]
+  - match: {timed_out: false}
+  - match: {hits.total.value: 2}
+  - match: {hits.total.relation: "eq"}
+  - match: {hits.sequences.0.join_keys.0: "SYSTEM"}
+  - match: {hits.sequences.0.events.0._id: "1"}
+  - match: {hits.sequences.0.events.0.fields.@timestamp: ["1580733296000"]}
+  - match: {hits.sequences.0.events.0.fields.id: [123]}
+  - match: {hits.sequences.0.events.0.fields.valid: [false]}
+  - match: {hits.sequences.0.events.0.fields.day_of_week: ["Monday"]}
+  - match: {hits.sequences.0.events.1._id: "2"}
+  - match: {hits.sequences.0.events.1.fields.@timestamp: ["1580819696000"]}
+  - match: {hits.sequences.0.events.1.fields.id: [123]}
+  - match: {hits.sequences.0.events.1.fields.valid: [true]}
+  - match: {hits.sequences.0.events.1.fields.day_of_week: ["Tuesday"]}
+  - match: {hits.sequences.1.join_keys.0: "SYSTEM"}
+  - match: {hits.sequences.1.events.0._id: "2"}
+  - match: {hits.sequences.1.events.0.fields.@timestamp: ["1580819696000"]}
+  - match: {hits.sequences.1.events.0.fields.id: [123]}
+  - match: {hits.sequences.1.events.0.fields.valid: [true]}
+  - match: {hits.sequences.1.events.0.fields.day_of_week: ["Tuesday"]}
+  - match: {hits.sequences.1.events.1._id: "3"}
+  - match: {hits.sequences.1.events.1.fields.@timestamp: ["1580906096000"]}
+  - match: {hits.sequences.1.events.1.fields.id: [123]}
+  - match: {hits.sequences.1.events.1.fields.valid: [true]}
+  - match: {hits.sequences.1.events.1.fields.day_of_week: ["Wednesday"]}
+
+---
+"Execute EQL sequence with filter_path":
+  - do:
+      eql.search:
+        index: eql_test
+        filter_path: "hits.sequences.join_keys,hits.sequences.events.fields.valid"
+        body:
+          query: 'sequence by user [process where user == "SYSTEM"] [process where true]'
+          fields: ["id","valid"]
+  - match: {hits.sequences.0.join_keys.0: "SYSTEM"}
+  - match: {hits.sequences.0.events.0.fields.valid: [false]}
+  - match: {hits.sequences.0.events.1.fields.valid: [true]}
+  - match: {hits.sequences.1.join_keys.0: "SYSTEM"}
+  - match: {hits.sequences.1.events.0.fields.valid: [true]}
+  - match: {hits.sequences.1.events.1.fields.valid: [true]}
+
 ---
 "Execute some EQL in async mode.":
   - do:

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/action/EqlSearchRequest.java

@@ -17,16 +17,22 @@ import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.ObjectParser;
+import org.elasticsearch.common.xcontent.ObjectParser.ValueType;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.XContentParser.Token;
 import org.elasticsearch.index.query.AbstractQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.elasticsearch.search.fetch.subphase.FieldAndFormat;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskId;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.function.Supplier;
@@ -51,6 +57,7 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
     private int fetchSize = RequestDefaults.FETCH_SIZE;
     private String query;
     private String resultPosition = "tail";
+    private List<FieldAndFormat> fetchFields;
 
     // Async settings
     private TimeValue waitForCompletionTimeout = null;
@@ -68,6 +75,7 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
     static final String KEY_KEEP_ALIVE = "keep_alive";
     static final String KEY_KEEP_ON_COMPLETION = "keep_on_completion";
     static final String KEY_RESULT_POSITION = "result_position";
+    static final String KEY_FETCH_FIELDS = "fields";
 
     static final ParseField FILTER = new ParseField(KEY_FILTER);
     static final ParseField TIMESTAMP_FIELD = new ParseField(KEY_TIMESTAMP_FIELD);
@@ -80,6 +88,7 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
     static final ParseField KEEP_ALIVE = new ParseField(KEY_KEEP_ALIVE);
     static final ParseField KEEP_ON_COMPLETION = new ParseField(KEY_KEEP_ON_COMPLETION);
     static final ParseField RESULT_POSITION = new ParseField(KEY_RESULT_POSITION);
+    static final ParseField FETCH_FIELDS_FIELD = SearchSourceBuilder.FETCH_FIELDS_FIELD;
 
     private static final ObjectParser<EqlSearchRequest, Void> PARSER = objectParser(EqlSearchRequest::new);
 
@@ -106,6 +115,11 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
         if (in.getVersion().onOrAfter(Version.V_7_10_0)) {
             resultPosition = in.readString();
         }
+        if (in.getVersion().onOrAfter(Version.V_7_12_0)) {
+            if (in.readBoolean()) {
+                fetchFields = in.readList(FieldAndFormat::new);
+            }
+        }
     }
 
     @Override
@@ -176,6 +190,9 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
         }
         builder.field(KEY_KEEP_ON_COMPLETION, keepOnCompletion);
         builder.field(KEY_RESULT_POSITION, resultPosition);
+        if (fetchFields != null && fetchFields.isEmpty() == false) {
+            builder.field(KEY_FETCH_FIELDS, fetchFields);
+        }
 
         return builder;
     }
@@ -201,6 +218,7 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
             (p, c) -> TimeValue.parseTimeValue(p.text(), KEY_KEEP_ALIVE), KEEP_ALIVE, ObjectParser.ValueType.VALUE);
         parser.declareBoolean(EqlSearchRequest::keepOnCompletion, KEEP_ON_COMPLETION);
         parser.declareString(EqlSearchRequest::resultPosition, RESULT_POSITION);
+        parser.declareField(EqlSearchRequest::fetchFields, EqlSearchRequest::parseFetchFields, FETCH_FIELDS_FIELD, ValueType.VALUE_ARRAY);
         return parser;
     }
 
@@ -303,6 +321,27 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
         return this;
     }
 
+    public List<FieldAndFormat> fetchFields() {
+        return fetchFields;
+    }
+
+    public EqlSearchRequest fetchFields(List<FieldAndFormat> fetchFields) {
+        this.fetchFields = fetchFields;
+        return this;
+    }
+
+    private static List<FieldAndFormat> parseFetchFields(XContentParser parser) throws IOException {
+        List<FieldAndFormat> result = new ArrayList<>();
+        Token token = parser.currentToken();
+
+        if (token == Token.START_ARRAY) {
+            while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) {
+                result.add(FieldAndFormat.fromXContent(parser));
+            }
+        }
+        return result.isEmpty() ? null : result;
+    }
+
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
@@ -324,6 +363,12 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
         if (out.getVersion().onOrAfter(Version.V_7_10_0)) { // TODO: Remove after backport
             out.writeString(resultPosition);
         }
+        if (out.getVersion().onOrAfter(Version.V_7_12_0)) {
+            out.writeBoolean(fetchFields != null);
+            if (fetchFields != null) {
+                out.writeList(fetchFields);
+            }
+        }
     }
 
     @Override
@@ -346,7 +391,8 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
                 Objects.equals(query, that.query) &&
                 Objects.equals(waitForCompletionTimeout, that.waitForCompletionTimeout) &&
                 Objects.equals(keepAlive, that.keepAlive) &&
-                Objects.equals(resultPosition, that.resultPosition);
+                Objects.equals(resultPosition, that.resultPosition) &&
+                Objects.equals(fetchFields, that.fetchFields);
     }
 
 
@@ -364,7 +410,8 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
             query,
             waitForCompletionTimeout,
             keepAlive,
-            resultPosition);
+            resultPosition,
+            fetchFields);
     }
 
     @Override

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/action/EqlSearchResponse.java

@@ -7,11 +7,13 @@
 package org.elasticsearch.xpack.eql.action;
 
 import org.apache.lucene.search.TotalHits;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.document.DocumentField;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
@@ -31,7 +33,9 @@ import org.elasticsearch.search.SearchHits;
 
 import java.io.IOException;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 
 import static org.elasticsearch.common.xcontent.ConstructingObjectParser.constructorArg;
@@ -190,15 +194,18 @@ public class EqlSearchResponse extends ActionResponse implements ToXContentObjec
             static final String INDEX = GetResult._INDEX;
             static final String ID = GetResult._ID;
             static final String SOURCE = SourceFieldMapper.NAME;
+            static final String FIELDS = "fields";
         }
 
         private static final ParseField INDEX = new ParseField(Fields.INDEX);
         private static final ParseField ID = new ParseField(Fields.ID);
         private static final ParseField SOURCE = new ParseField(Fields.SOURCE);
+        private static final ParseField FIELDS = new ParseField(Fields.FIELDS);
 
+        @SuppressWarnings("unchecked")
         private static final ConstructingObjectParser<Event, Void> PARSER =
-                new ConstructingObjectParser<>("eql/search_response_event", true,
-                        args -> new Event((String) args[0], (String) args[1], (BytesReference) args[2]));
+                new ConstructingObjectParser<>("eql/search_response_event", true, 
+                    args -> new Event((String) args[0], (String) args[1], (BytesReference) args[2], (Map<String, DocumentField>) args[3]));
 
         static {
             PARSER.declareString(constructorArg(), INDEX);
@@ -209,22 +216,37 @@ public class EqlSearchResponse extends ActionResponse implements ToXContentObjec
                     return BytesReference.bytes(builder);
                 }
             }, SOURCE);
+            PARSER.declareObject(optionalConstructorArg(), (p, c) -> {
+                Map<String, DocumentField> fields = new HashMap<>();
+                while (p.nextToken() != XContentParser.Token.END_OBJECT) {
+                    DocumentField field = DocumentField.fromXContent(p);
+                    fields.put(field.getName(), field);
+                }
+                return fields;
+            }, FIELDS);
         }
 
         private final String index;
         private final String id;
         private final BytesReference source;
+        private final Map<String, DocumentField> fetchFields;
 
-        public Event(String index, String id, BytesReference source) {
+        public Event(String index, String id, BytesReference source, Map<String, DocumentField> fetchFields) {
             this.index = index;
             this.id = id;
             this.source = source;
+            this.fetchFields = fetchFields;
         }
 
         public Event(StreamInput in) throws IOException {
             index = in.readString();
             id = in.readString();
             source = in.readBytesReference();
+            if (in.getVersion().onOrAfter(Version.V_7_12_0) && in.readBoolean()) {
+                fetchFields = in.readMap(StreamInput::readString, DocumentField::new);
+            } else {
+                fetchFields = null;
+            }
         }
 
         @Override
@@ -232,6 +254,12 @@ public class EqlSearchResponse extends ActionResponse implements ToXContentObjec
             out.writeString(index);
             out.writeString(id);
             out.writeBytesReference(source);
+            if (out.getVersion().onOrAfter(Version.V_7_12_0)) {
+                out.writeBoolean(fetchFields != null);
+                if (fetchFields != null) {
+                    out.writeMap(fetchFields, StreamOutput::writeString, (stream, documentField) -> documentField.writeTo(stream));
+                }
+            }
         }
 
         @Override
@@ -241,6 +269,17 @@ public class EqlSearchResponse extends ActionResponse implements ToXContentObjec
             builder.field(Fields.ID, id);
             // We have to use the deprecated version since we don't know the content type of the original source
             XContentHelper.writeRawField(Fields.SOURCE, source, builder, params);
+            // ignore fields all together if they are all empty
+            if (fetchFields != null && fetchFields.isEmpty() == false
+                && fetchFields.values().stream().anyMatch(df -> df.getValues().size() > 0)) {
+                builder.startObject(Fields.FIELDS);
+                for (DocumentField field : fetchFields.values()) {
+                    if (field.getValues().size() > 0) {
+                        field.toXContent(builder, params);
+                    }
+                }
+                builder.endObject();
+            }
             builder.endObject();
             return builder;
         }
@@ -261,9 +300,13 @@ public class EqlSearchResponse extends ActionResponse implements ToXContentObjec
             return source;
         }
 
+        public Map<String, DocumentField> fetchFields() {
+            return fetchFields;
+        }
+
         @Override
         public int hashCode() {
-            return Objects.hash(index, id, source);
+            return Objects.hash(index, id, source, fetchFields);
         }
 
         @Override
@@ -279,7 +322,8 @@ public class EqlSearchResponse extends ActionResponse implements ToXContentObjec
             EqlSearchResponse.Event other = (EqlSearchResponse.Event) obj;
             return Objects.equals(index, other.index)
                     && Objects.equals(id, other.id)
-                    && Objects.equals(source, other.source);
+                    && Objects.equals(source, other.source)
+                    && Objects.equals(fetchFields, other.fetchFields);
         }
 
         @Override

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/execution/payload/EventPayload.java

@@ -25,7 +25,7 @@ public class EventPayload extends AbstractPayload {
         List<SearchHit> hits = RuntimeUtils.searchHits(response);
         values = new ArrayList<>(hits.size());
         for (SearchHit hit : hits) {
-            values.add(new Event(hit.getIndex(), hit.getId(), hit.getSourceRef()));
+            values.add(new Event(hit.getIndex(), hit.getId(), hit.getSourceRef(), hit.getFields()));
         }
     }
 

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/execution/search/BasicQueryClient.java

@@ -19,6 +19,7 @@ import org.elasticsearch.index.query.IdsQueryBuilder;
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
+import org.elasticsearch.search.fetch.subphase.FieldAndFormat;
 import org.elasticsearch.tasks.TaskCancelledException;
 import org.elasticsearch.xpack.eql.EqlIllegalArgumentException;
 import org.elasticsearch.xpack.eql.session.EqlConfiguration;
@@ -44,11 +45,13 @@ public class BasicQueryClient implements QueryClient {
     final EqlConfiguration cfg;
     final Client client;
     final String[] indices;
+    final List<FieldAndFormat> fetchFields;
 
     public BasicQueryClient(EqlSession eqlSession) {
         this.cfg = eqlSession.configuration();
         this.client = eqlSession.client();
         this.indices = cfg.indices();
+        this.fetchFields = cfg.fetchFields();
     }
 
     @Override
@@ -137,6 +140,9 @@ public class BasicQueryClient implements QueryClient {
                 // the default size is 10 so be sure to change it
                 // NB:this is different from mget
                 .size(idQuery.ids().size());
+            if (fetchFields != null) {
+                fetchFields.forEach(builder::fetchField);
+            }
 
             SearchRequest search = prepareRequest(builder, false, entry.getKey());
             multiSearchBuilder.add(search);

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/execution/search/SourceGenerator.java

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.eql.execution.search;
 
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.elasticsearch.search.fetch.subphase.FieldAndFormat;
 import org.elasticsearch.search.sort.FieldSortBuilder;
 import org.elasticsearch.search.sort.NestedSortBuilder;
 import org.elasticsearch.search.sort.ScriptSortBuilder.ScriptSortType;
@@ -20,6 +21,8 @@ import org.elasticsearch.xpack.ql.querydsl.container.AttributeSort;
 import org.elasticsearch.xpack.ql.querydsl.container.ScriptSort;
 import org.elasticsearch.xpack.ql.querydsl.container.Sort;
 
+import java.util.List;
+
 import static org.elasticsearch.index.query.QueryBuilders.boolQuery;
 import static org.elasticsearch.search.sort.SortBuilders.fieldSort;
 import static org.elasticsearch.search.sort.SortBuilders.scriptSort;
@@ -28,7 +31,7 @@ public abstract class SourceGenerator {
 
     private SourceGenerator() {}
 
-    public static SearchSourceBuilder sourceBuilder(QueryContainer container, QueryBuilder filter) {
+    public static SearchSourceBuilder sourceBuilder(QueryContainer container, QueryBuilder filter, List<FieldAndFormat> fetchFields) {
         QueryBuilder finalQuery = null;
         // add the source
         if (container.query() != null) {
@@ -61,6 +64,11 @@ public abstract class SourceGenerator {
         // disable the source, as we rely on "fields" API
         source.fetchSource(false);
 
+        // add the "fields" to be fetched
+        if (fetchFields != null) {
+            fetchFields.forEach(source::fetchField);
+        }
+
         if (container.limit() != null) {
             // add size and from
             source.size(container.limit().absLimit());

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/execution/sequence/SequencePayload.java

@@ -28,7 +28,7 @@ class SequencePayload extends AbstractPayload {
             List<SearchHit> hits = docs.get(i);
             List<Event> events = new ArrayList<>(hits.size());
             for (SearchHit hit : hits) {
-                events.add(new Event(hit.getIndex(), hit.getId(), hit.getSourceRef()));
+                events.add(new Event(hit.getIndex(), hit.getId(), hit.getSourceRef(), hit.getFields()));
             }
             values.add(new org.elasticsearch.xpack.eql.action.EqlSearchResponse.Sequence(s.key().asList(), events));
         }

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/plan/physical/EsQueryExec.java

@@ -55,7 +55,7 @@ public class EsQueryExec extends LeafExec {
     public SearchSourceBuilder source(EqlSession session) {
         EqlConfiguration cfg = session.configuration();
         // by default use the configuration size
-        return SourceGenerator.sourceBuilder(queryContainer, cfg.filter());
+        return SourceGenerator.sourceBuilder(queryContainer, cfg.filter(), cfg.fetchFields());
     }
 
     @Override

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/plugin/TransportEqlSearchAction.java

@@ -21,6 +21,7 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.time.DateUtils;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.search.fetch.subphase.FieldAndFormat;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -42,6 +43,7 @@ import org.elasticsearch.xpack.ql.expression.Order;
 
 import java.io.IOException;
 import java.time.ZoneId;
+import java.util.List;
 import java.util.Map;
 
 import static org.elasticsearch.action.ActionListener.wrap;
@@ -117,6 +119,7 @@ public class TransportEqlSearchAction extends HandledTransportAction<EqlSearchRe
         // TODO: these should be sent by the client
         ZoneId zoneId = DateUtils.of("Z");
         QueryBuilder filter = request.filter();
+        List<FieldAndFormat> fetchFields = request.fetchFields();
         TimeValue timeout = TimeValue.timeValueSeconds(30);
         String clientId = null;
 
@@ -128,7 +131,7 @@ public class TransportEqlSearchAction extends HandledTransportAction<EqlSearchRe
             .size(request.size())
             .fetchSize(request.fetchSize());
 
-        EqlConfiguration cfg = new EqlConfiguration(request.indices(), zoneId, username, clusterName, filter, timeout,
+        EqlConfiguration cfg = new EqlConfiguration(request.indices(), zoneId, username, clusterName, filter, fetchFields, timeout,
                 request.indicesOptions(), request.fetchSize(), clientId, new TaskId(nodeId, task.getId()), task);
         executeRequestWithRetryAttempt(clusterService, listener::onFailure,
             onFailure -> planExecutor.eql(cfg, request.query(), params,

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/querydsl/container/QueryContainer.java

@@ -169,7 +169,7 @@ public class QueryContainer {
     public String toString() {
         try (XContentBuilder builder = JsonXContent.contentBuilder()) {
             builder.humanReadable(true).prettyPrint();
-            SourceGenerator.sourceBuilder(this, null).toXContent(builder, ToXContent.EMPTY_PARAMS);
+            SourceGenerator.sourceBuilder(this, null, null).toXContent(builder, ToXContent.EMPTY_PARAMS);
             return Strings.toString(builder);
         } catch (IOException e) {
             throw new EqlIllegalArgumentException("error rendering", e);

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/session/EqlConfiguration.java

@@ -12,10 +12,12 @@ import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.search.fetch.subphase.FieldAndFormat;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.xpack.eql.action.EqlSearchTask;
 
 import java.time.ZoneId;
+import java.util.List;
 
 public class EqlConfiguration extends org.elasticsearch.xpack.ql.session.Configuration {
 
@@ -29,14 +31,17 @@ public class EqlConfiguration extends org.elasticsearch.xpack.ql.session.Configu
 
     @Nullable
     private final QueryBuilder filter;
+    @Nullable
+    private final List<FieldAndFormat> fetchFields;
 
-    public EqlConfiguration(String[] indices, ZoneId zi, String username, String clusterName, QueryBuilder filter, TimeValue requestTimeout,
-                            IndicesOptions indicesOptions, int fetchSize, String clientId, TaskId taskId,
-                            EqlSearchTask task) {
+    public EqlConfiguration(String[] indices, ZoneId zi, String username, String clusterName, QueryBuilder filter,
+                            List<FieldAndFormat> fetchFields, TimeValue requestTimeout, IndicesOptions indicesOptions, int fetchSize,
+                            String clientId, TaskId taskId, EqlSearchTask task) {
         super(zi, username, clusterName);
 
         this.indices = indices;
         this.filter = filter;
+        this.fetchFields = fetchFields;
         this.requestTimeout = requestTimeout;
         this.clientId = clientId;
         this.indicesOptions = indicesOptions;
@@ -65,6 +70,10 @@ public class EqlConfiguration extends org.elasticsearch.xpack.ql.session.Configu
         return filter;
     }
 
+    public List<FieldAndFormat> fetchFields() {
+        return fetchFields;
+    }
+
     public String clientId() {
         return clientId;
     }

x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/AbstractBWCWireSerializingTestCase.java

@@ -0,0 +1,52 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+package org.elasticsearch.xpack.eql;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.elasticsearch.Version.getDeclaredVersions;
+import static org.hamcrest.Matchers.equalTo;
+
+public abstract class AbstractBWCWireSerializingTestCase<T extends Writeable> extends AbstractWireSerializingTestCase<T> {
+
+    private static final List<Version> ALL_VERSIONS = Collections.unmodifiableList(getDeclaredVersions(Version.class));
+    private static Version EQL_GA_VERSION = Version.V_7_10_0;
+
+    private static List<Version> getAllBWCVersions(Version version) {
+        return ALL_VERSIONS.stream().filter(v -> v.onOrAfter(EQL_GA_VERSION) && v.before(version) && version.isCompatible(v)).collect(
+            Collectors.toList());
+    }
+
+    private static final List<Version> DEFAULT_BWC_VERSIONS = getAllBWCVersions(Version.CURRENT);
+
+    public final void testBwcSerialization() throws IOException {
+        for (int runs = 0; runs < NUMBER_OF_TEST_RUNS; runs++) {
+            T testInstance = createTestInstance();
+            for (Version bwcVersion : DEFAULT_BWC_VERSIONS) {
+                assertBwcSerialization(testInstance, bwcVersion);
+            }
+        }
+    }
+
+    protected final void assertBwcSerialization(T testInstance, Version version) throws IOException {
+        T deserializedInstance = copyInstance(testInstance, version);
+        assertOnBWCObject(testInstance, deserializedInstance, version);
+    }
+
+    protected void assertOnBWCObject(T testInstance, T bwcDeserializedObject, Version version) {
+        assertNotSame(version.toString(), bwcDeserializedObject, testInstance);
+        assertThat(version.toString(), testInstance, equalTo(bwcDeserializedObject));
+        assertEquals(version.toString(), testInstance.hashCode(), bwcDeserializedObject.hashCode());
+    }
+}

x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/EqlTestUtils.java

@@ -36,7 +36,7 @@ public final class EqlTestUtils {
     }
 
     public static final EqlConfiguration TEST_CFG = new EqlConfiguration(new String[] {"none"},
-            org.elasticsearch.xpack.ql.util.DateUtils.UTC, "nobody", "cluster", null, TimeValue.timeValueSeconds(30), null,
+            org.elasticsearch.xpack.ql.util.DateUtils.UTC, "nobody", "cluster", null, null, TimeValue.timeValueSeconds(30), null,
             123, "", new TaskId("test", 123), null);
 
     public static EqlConfiguration randomConfiguration() {
@@ -45,6 +45,7 @@ public final class EqlTestUtils {
             randomAlphaOfLength(16),
             randomAlphaOfLength(16),
             null,
+            null,
             new TimeValue(randomNonNegativeLong()),
             randomIndicesOptions(),
             randomIntBetween(1, 1000),
@@ -70,5 +71,4 @@ public final class EqlTestUtils {
         return IndicesOptions.fromOptions(randomBoolean(), randomBoolean(), randomBoolean(), randomBoolean(), randomBoolean(),
             randomBoolean(), randomBoolean(), randomBoolean(), randomBoolean());
     }
-
 }

x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/action/EqlSearchRequestTests.java

@@ -9,20 +9,19 @@ package org.elasticsearch.xpack.eql.action;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.text.Text;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.search.SearchModule;
-import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.search.fetch.subphase.FieldAndFormat;
 import org.elasticsearch.xpack.eql.AbstractBWCSerializationTestCase;
 import org.junit.Before;
 
 import java.io.IOException;
-import java.util.Arrays;
+import java.util.ArrayList;
 import java.util.Collections;
-import java.util.function.Supplier;
+import java.util.List;
 
 import static org.elasticsearch.index.query.AbstractQueryBuilder.parseInnerQueryBuilder;
 
@@ -56,6 +55,14 @@ public class EqlSearchRequestTests extends AbstractBWCSerializationTestCase<EqlS
     @Override
     protected EqlSearchRequest createTestInstance() {
         try {
+            List<FieldAndFormat> randomFetchFields = new ArrayList<>();
+            int fetchFieldsCount = randomIntBetween(0, 5);
+            for (int j = 0; j < fetchFieldsCount; j++) {
+                randomFetchFields.add(new FieldAndFormat(randomAlphaOfLength(10), randomAlphaOfLength(10)));
+            }
+            if (randomFetchFields.isEmpty()) {
+                randomFetchFields = null;
+            }
             QueryBuilder filter = parseFilter(defaultTestFilter);
             EqlSearchRequest request = new EqlSearchRequest()
                 .indices(new String[]{defaultTestIndex})
@@ -64,7 +71,8 @@ public class EqlSearchRequestTests extends AbstractBWCSerializationTestCase<EqlS
                 .eventCategoryField(randomAlphaOfLength(10))
                 .fetchSize(randomIntBetween(1, 50))
                 .size(randomInt(50))
-                .query(randomAlphaOfLength(10));
+                .query(randomAlphaOfLength(10))
+                .fetchFields(randomFetchFields);
 
             return request;
         } catch (IOException ex) {
@@ -84,21 +92,6 @@ public class EqlSearchRequestTests extends AbstractBWCSerializationTestCase<EqlS
         return parseInnerQueryBuilder;
     }
 
-    private Object randomValue() {
-        Supplier<Object> value = randomFrom(Arrays.asList(
-            ESTestCase::randomInt,
-            ESTestCase::randomFloat,
-            ESTestCase::randomLong,
-            ESTestCase::randomDouble,
-            () -> randomAlphaOfLengthBetween(5, 20),
-            ESTestCase::randomBoolean,
-            ESTestCase::randomByte,
-            ESTestCase::randomShort,
-            () -> new Text(randomAlphaOfLengthBetween(5, 20)),
-            () -> null));
-        return value.get();
-    }
-
     @Override
     protected Writeable.Reader<EqlSearchRequest> instanceReader() {
         return EqlSearchRequest::new;

x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/action/EqlSearchResponseTests.java

@@ -8,23 +8,44 @@ package org.elasticsearch.xpack.eql.action;
 
 import org.apache.lucene.search.TotalHits;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.document.DocumentField;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
-import org.elasticsearch.xpack.eql.AbstractBWCSerializationTestCase;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.RandomObjects;
+import org.elasticsearch.xpack.eql.AbstractBWCWireSerializingTestCase;
 import org.elasticsearch.xpack.eql.action.EqlSearchResponse.Event;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.function.Supplier;
 
-public class EqlSearchResponseTests extends AbstractBWCSerializationTestCase<EqlSearchResponse> {
+import static org.elasticsearch.common.xcontent.XContentHelper.toXContent;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertToXContentEquivalent;
+
+public class EqlSearchResponseTests extends AbstractBWCWireSerializingTestCase<EqlSearchResponse> {
+
+    public void testFromXContent() throws IOException {
+        XContentType xContentType = randomFrom(XContentType.values()).canonical();
+        EqlSearchResponse response = randomEqlSearchResponse(xContentType);
+        boolean humanReadable = randomBoolean();
+        BytesReference originalBytes = toShuffledXContent(response, xContentType, ToXContent.EMPTY_PARAMS, humanReadable);
+        EqlSearchResponse parsed;
+        try (XContentParser parser = createParser(xContentType.xContent(), originalBytes)) {
+            parsed = EqlSearchResponse.fromXContent(parser);
+        }
+        assertToXContentEquivalent(originalBytes, toXContent(parsed, xContentType, humanReadable), xContentType);
+    }
 
     private static class RandomSource implements ToXContentObject {
 
@@ -78,7 +99,14 @@ public class EqlSearchResponseTests extends AbstractBWCSerializationTestCase<Eql
             hits = new ArrayList<>();
             for (int i = 0; i < size; i++) {
                 BytesReference bytes = new RandomSource(() -> randomAlphaOfLength(10)).toBytes(xType);
-                hits.add(new Event(String.valueOf(i), randomAlphaOfLength(10), bytes));
+                Map<String, DocumentField> fetchFields = new HashMap<>();
+                for (int j = 0; j < randomIntBetween(0, 5); j++) {
+                    fetchFields.put(randomAlphaOfLength(10), randomDocumentField(xType).v1());
+                }
+                if (fetchFields.isEmpty() && randomBoolean()) {
+                    fetchFields = null;
+                }
+                hits.add(new Event(String.valueOf(i), randomAlphaOfLength(10), bytes, fetchFields));
             }
         }
         if (randomBoolean()) {
@@ -87,9 +115,28 @@ public class EqlSearchResponseTests extends AbstractBWCSerializationTestCase<Eql
         return null;
     }
 
-    @Override
-    protected EqlSearchResponse createXContextTestInstance(XContentType xContentType) {
-        return randomEqlSearchResponse(xContentType);
+    private static Tuple<DocumentField, DocumentField> randomDocumentField(XContentType xType) {
+        switch (randomIntBetween(0, 2)) {
+            case 0:
+                String fieldName = randomAlphaOfLengthBetween(3, 10);
+                Tuple<List<Object>, List<Object>> tuple = RandomObjects.randomStoredFieldValues(random(), xType);
+                DocumentField input = new DocumentField(fieldName, tuple.v1());
+                DocumentField expected = new DocumentField(fieldName, tuple.v2());
+                return Tuple.tuple(input, expected);
+            case 1:
+                List<Object> listValues = randomList(1, 5, () -> randomList(1, 5, ESTestCase::randomInt));
+                DocumentField listField = new DocumentField(randomAlphaOfLength(5), listValues);
+                return Tuple.tuple(listField, listField);
+            case 2:
+                List<Object> objectValues = randomList(1, 5, () ->
+                    Map.of(randomAlphaOfLength(5), randomInt(),
+                        randomAlphaOfLength(5), randomBoolean(),
+                        randomAlphaOfLength(5), randomAlphaOfLength(10)));
+                DocumentField objectField = new DocumentField(randomAlphaOfLength(5), objectValues);
+                return Tuple.tuple(objectField, objectField);
+            default:
+                throw new IllegalStateException();
+        }
     }
 
     @Override
@@ -170,9 +217,5 @@ public class EqlSearchResponseTests extends AbstractBWCSerializationTestCase<Eql
                 return null;
         }
     }
-
-    @Override
-    protected EqlSearchResponse doParseInstance(XContentParser parser) {
-        return EqlSearchResponse.fromXContent(parser);
-    }
+    
 }