EQL: multi-value fields support

x-pack/plugin/eql/qa/mixed-node/src/test/java/org/elasticsearch/xpack/eql/qa/mixed_node/EqlSearchIT.java

@@ -8,6 +8,7 @@
 package org.elasticsearch.xpack.eql.qa.mixed_node;
 
 import org.apache.http.HttpHost;
+import org.elasticsearch.Version;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.RestClient;
@@ -17,17 +18,23 @@
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.test.NotEqualMessageBuilder;
 import org.elasticsearch.test.rest.ESRestTestCase;
+import org.elasticsearch.xpack.eql.execution.search.RuntimeUtils;
+import org.elasticsearch.xpack.eql.expression.function.EqlFunctionRegistry;
 import org.elasticsearch.xpack.ql.TestNode;
 import org.elasticsearch.xpack.ql.TestNodes;
+import org.elasticsearch.xpack.ql.expression.function.FunctionDefinition;
 import org.junit.After;
 import org.junit.Before;
 
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 import static java.util.Arrays.asList;
 import static java.util.Collections.emptyMap;
@@ -94,6 +101,91 @@ public void testSequencesWithRequestToUpgradedNodes() throws Exception {
         assertSequncesQueryOnNodes(newNodes);
     }
 
+    /**
+     * Requests are sent to the new (upgraded) version nodes of the cluster. The request should be redirected to the old nodes at this point
+     * if their version is lower than {@code org.elasticsearch.xpack.eql.execution.search.RuntimeUtils.SWITCH_TO_MULTI_VALUE_FIELDS_VERSION}
+     * version.
+     */
+    public void testMultiValueFields() throws Exception {
+        final String bulkEntries = readResource(EqlSearchIT.class.getResourceAsStream("/eql_data.json"));
+        Request bulkRequst = new Request("POST", index + "/_bulk?refresh");
+        bulkRequst.setJsonEntity(bulkEntries);
+        assertOK(client().performRequest(bulkRequst));
+
+        // build a set of functions names to check if all functions are tested with multi-value fields
+        final Set<String> availableFunctions = new EqlFunctionRegistry().listFunctions()
+            .stream()
+            .map(FunctionDefinition::name)
+            .collect(Collectors.toSet());
+        // each function has a query and query results associated to it
+        Set<String> testedFunctions = new HashSet<>();
+        // TODO: remove the 8.0.0 version check after code reaches 7.x as well
+        boolean multiValued = newNodes.get(0).getVersion() != Version.V_8_0_0
+            && nodes.getBWCVersion().onOrAfter(RuntimeUtils.SWITCH_TO_MULTI_VALUE_FIELDS_VERSION);
+        try (
+            // TODO: use newNodes (instead of bwcNodes) after code reaches 7.x as well
+            RestClient client = buildClient(restClientSettings(),
+                bwcNodes.stream().map(TestNode::getPublishAddress).toArray(HttpHost[]::new))
+        ) {
+            // filter only the relevant bits of the response
+            String filterPath = "filter_path=hits.events._id";
+            Request request = new Request("POST", index + "/_eql/search?" + filterPath);
+
+            assertMultiValueFunctionQuery(availableFunctions, testedFunctions, request, client, "between",
+                "PROCESS where between(process_name, \\\"w\\\", \\\"s\\\") : \\\"indow\\\"",
+                multiValued ? new int[] {120, 121} : new int[] {121});
+            assertMultiValueFunctionQuery(availableFunctions, testedFunctions, request, client, "cidrmatch",
+                "PROCESS where string(cidrmatch(source_address, \\\"10.6.48.157/24\\\")) : \\\"true\\\"",
+                multiValued ? new int[] {121, 122} : new int[] {122});
+            assertMultiValueFunctionQuery(availableFunctions, testedFunctions, request, client, "concat",
+                "PROCESS where concat(file_name, process_name) == \\\"foo\\\" or add(pid, ppid) > 100",
+                multiValued ? new int[] {116, 117, 120, 121, 122} : new int[] {120, 121});
+            assertMultiValueFunctionQuery(availableFunctions, testedFunctions, request, client, "endswith",
+                "PROCESS where string(endswith(process_name, \\\"s\\\")) : \\\"true\\\"",
+                multiValued ? new int[] {120, 121} : new int[] {121});
+            assertMultiValueFunctionQuery(availableFunctions, testedFunctions, request, client, "indexof",
+                "PROCESS where indexof(file_name, \\\"x\\\", 2) > 0",
+                multiValued ? new int[] {116, 117} : new int[] {117});
+            assertMultiValueFunctionQuery(availableFunctions, testedFunctions, request, client, "length",
+                "PROCESS where length(file_name) >= 3 and length(file_name) == 1",
+                multiValued ? new int[] {116} : new int[] {});
+            assertMultiValueFunctionQuery(availableFunctions, testedFunctions, request, client, "startswith",
+                "PROCESS where string(startswith~(file_name, \\\"F\\\")) : \\\"true\\\"",
+                multiValued ? new int[] {116, 117, 120, 121} : new int[] {116, 120, 121});
+            assertMultiValueFunctionQuery(availableFunctions, testedFunctions, request, client, "string",
+                "PROCESS where string(concat(file_name, process_name) == \\\"foo\\\") : \\\"true\\\"",
+                multiValued ? new int[] {116, 120} : new int[] {120});
+            assertMultiValueFunctionQuery(availableFunctions, testedFunctions, request, client, "stringcontains",
+                "PROCESS where string(stringcontains(file_name, \\\"txt\\\")) : \\\"true\\\"",
+                multiValued ? new int[] {117} : new int[] {});
+            assertMultiValueFunctionQuery(availableFunctions, testedFunctions, request, client, "substring",
+                "PROCESS where substring(file_name, -4) : \\\".txt\\\"",
+                multiValued ? new int[] {117} : new int[] {});
+            assertMultiValueFunctionQuery(availableFunctions, testedFunctions, request, client, "add",
+                "PROCESS where add(pid, 1) == 2",
+                multiValued ? new int[] {120, 121, 122} : new int[] {120, 121, 122});
+            assertMultiValueFunctionQuery(availableFunctions, testedFunctions, request, client, "divide",
+                "PROCESS where divide(pid, 12) == 1",
+                multiValued ? new int[] {116, 117, 118, 119, 120, 122} : new int[] {116, 117, 118, 119});
+            assertMultiValueFunctionQuery(availableFunctions, testedFunctions, request, client, "modulo",
+                "PROCESS where modulo(ppid, 10) == 0",
+                multiValued ? new int[] {121, 122} : new int[] {121});
+            assertMultiValueFunctionQuery(availableFunctions, testedFunctions, request, client, "multiply",
+                "PROCESS where multiply(pid, 10) == 120",
+                multiValued ? new int[] {116, 117, 118, 119, 120, 122} : new int[] {116, 117, 118, 119, 120, 122});
+            assertMultiValueFunctionQuery(availableFunctions, testedFunctions, request, client, "number",
+                "PROCESS where number(command_line) + pid >= 360",
+                multiValued ? new int[] {122, 123} : new int[] {123});
+            assertMultiValueFunctionQuery(availableFunctions, testedFunctions, request, client, "subtract",
+                "PROCESS where subtract(pid, 1) == 0",
+                multiValued ? new int[] {120, 121, 122} : new int[] {120, 121, 122});
+        }
+
+        // check that ALL functions from the function registry have a test query. We don't want to miss any of the functions, since this
+        // is about painless scripting
+        assertTrue(testedFunctions.containsAll(availableFunctions));
+    }
+
     private void assertEventsQueryOnNodes(List<TestNode> nodesList) throws Exception {
         final String event = randomEvent();
         Map<String, Object> expectedResponse = prepareEventsTestData(event);
@@ -222,6 +314,17 @@ private Map<String, Object> prepareSequencesTestData() throws IOException {
         return expectedResponse;
     }
 
+    private void assertMultiValueFunctionQuery(Set<String> availableFunctions, Set<String> testedFunctions, Request request,
+        RestClient client, String functionName, String query, int[] ids) throws IOException {
+        List<Object> eventIds = new ArrayList<>();
+        for (int id : ids) {
+            eventIds.add(String.valueOf(id));
+        }
+        request.setJsonEntity("{\"query\":\"" + query + "\"}");
+        assertResponse(query, eventIds, runEql(client, request));
+        testedFunctions.add(functionName);
+    }
+
     private void assertResponse(Map<String, Object> expected, Map<String, Object> actual) {
         if (false == expected.equals(actual)) {
             NotEqualMessageBuilder message = new NotEqualMessageBuilder();
@@ -230,6 +333,29 @@ private void assertResponse(Map<String, Object> expected, Map<String, Object> ac
         }
     }
 
+    @SuppressWarnings("unchecked")
+    private void assertResponse(String query, List<Object> expected, Map<String, Object> actual) {
+        List<Map<String, Object>> events = new ArrayList<>();
+        Map<String, Object> hits = (Map<String, Object>) actual.get("hits");
+        if (hits == null || hits.isEmpty()) {
+            if (expected.isEmpty()) {
+                return;
+            }
+            fail("For query [" + query + "]\nResponse does not match: the returned list of resuts is empty.\nExpected " + expected);
+        } else {
+            events = (List<Map<String, Object>>) hits.get("events");
+        }
+
+        List<Object> actualList = new ArrayList<>();
+        events.stream().forEach(m -> actualList.add(m.get("_id")));
+
+        if (false == expected.equals(actualList)) {
+            NotEqualMessageBuilder message = new NotEqualMessageBuilder();
+            message.compareLists(actualList, expected);
+            fail("For query [" + query + "]\nResponse does not match:\n" + message.toString());
+        }
+    }
+
     private Map<String, Object> runEql(RestClient client, Request request) throws IOException {
         Response response = client.performRequest(request);
         try (InputStream content = response.getEntity().getContent()) {

x-pack/plugin/eql/qa/mixed-node/src/test/resources/eql_data.json

@@ -28,3 +28,19 @@
 {"@timestamp":"12345678914","event_type":"success","sequence":44,"correlation_success1":"C","correlation_success2":"D"}
 {"index":{"_id":15}}
 {"@timestamp":"12345678999","event_type":"failure","sequence":44,"correlation_failure1":"C","correlation_failure2":"D"}
+{"index":{"_id":116}}
+{"@timestamp":"123456789116","event_type":"PROCESS","file_name":["x","f","zyx"],"process_name":["oo","abc"],"pid":[12,13,14],"ppid":1}
+{"index":{"_id":117}}
+{"@timestamp":"123456789117","event_type":"PROCESS","file_name":["a.exe","f.txt"],"process_name":"oo","pid":[12,13,14],"ppid":[89,1,2,3]}
+{"index":{"_id":118}}
+{"@timestamp":"123456789118","event_type":"PROCESS","file_name":"a","process_name":"oo","pid":12}
+{"index":{"_id":119}}
+{"@timestamp":"123456789119","event_type":"PROCESS","process_name":["oo","foo"],"pid":[121,12]}
+{"index":{"_id":120}}
+{"@timestamp":"123456789120","event_type":"PROCESS","file_name":["f","g","f"],"process_name":["oo","pp","windows"],"pid":[12,1,2,333],"ppid":121}
+{"index":{"_id":121}}
+{"@timestamp":"123456789121","event_type":"PROCESS","file_name":"f","pid":1,"ppid":[100,1000],"source_address":["127.0.0.1","10.6.48.157","10.0.0.5"],"process_name":"windows"}
+{"index":{"_id":122}}
+{"@timestamp":"123456789122","event_type":"PROCESS","pid":[1,2,3,4,5,6,12],"ppid":[66,67,68,69,99,100],"source_address":"10.6.48.157","command_line":"348"}
+{"index":{"_id":123}}
+{"@timestamp":"123456789123","event_type":"PROCESS","pid":500,"command_line":"100"}

x-pack/plugin/eql/qa/mixed-node/src/test/resources/eql_mapping.json

@@ -31,5 +31,23 @@
           "path": "sequence"
         }
       }
+    },
+    "command_line": {
+      "type": "keyword"
+    },
+    "file_name": {
+      "type": "keyword"
+    },
+    "process_name": {
+      "type": "keyword"
+    },
+    "pid" : {
+      "type" : "long"
+    },
+    "ppid" : {
+      "type" : "long"
+    },
+    "source_address" : {
+      "type" : "ip"
     }
   }

x-pack/plugin/eql/src/internalClusterTest/java/org/elasticsearch/xpack/eql/action/AsyncEqlSearchActionIT.java

@@ -332,8 +332,8 @@ public static class FakePainlessScriptPlugin extends MockScriptPlugin {
         @Override
         protected Map<String, Function<Map<String, Object>, Object>> pluginScripts() {
             Map<String, Function<Map<String, Object>, Object>> scripts = new HashMap<>();
-            scripts.put("InternalQlScriptUtils.nullSafeFilter(InternalQlScriptUtils.eq(InternalQlScriptUtils.div(" +
-                "params.v0,InternalQlScriptUtils.docValue(doc,params.v1)),params.v2))", FakePainlessScriptPlugin::fail);
+            scripts.put("InternalEqlScriptUtils.multiValueDocValues(doc,params.v0,X0 -> InternalQlScriptUtils.nullSafeFilter("
+                + "InternalQlScriptUtils.eq(InternalQlScriptUtils.div(params.v1,X0),params.v2)))", FakePainlessScriptPlugin::fail);
             return scripts;
         }
 

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

@@ -9,6 +9,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.search.MultiSearchResponse;
 import org.elasticsearch.action.search.SearchRequest;
@@ -41,11 +42,11 @@
 import java.util.Set;
 
 import static org.elasticsearch.index.query.QueryBuilders.boolQuery;
-import static org.elasticsearch.xpack.ql.execution.search.QlSourceBuilder.SWITCH_TO_FIELDS_API_VERSION;
 
 public final class RuntimeUtils {
 
     static final Logger QUERY_LOG = LogManager.getLogger(QueryClient.class);
+    public static final Version SWITCH_TO_MULTI_VALUE_FIELDS_VERSION = Version.V_7_15_0;
 
     private RuntimeUtils() {}
 
@@ -148,7 +149,7 @@ public static HitExtractor createExtractor(FieldExtraction ref, EqlConfiguration
     public static SearchRequest prepareRequest(SearchSourceBuilder source,
                                                boolean includeFrozen,
                                                String... indices) {
-        SearchRequest searchRequest = new SearchRequest(SWITCH_TO_FIELDS_API_VERSION);
+        SearchRequest searchRequest = new SearchRequest(SWITCH_TO_MULTI_VALUE_FIELDS_VERSION);
         searchRequest.indices(indices);
         searchRequest.source(source);
         searchRequest.allowPartialSearchResults(false);

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/expression/function/scalar/whitelist/InternalEqlScriptUtils.java

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.eql.expression.function.scalar.whitelist;
 
+import org.elasticsearch.index.fielddata.ScriptDocValues;
 import org.elasticsearch.xpack.eql.expression.function.scalar.math.ToNumberFunctionProcessor;
 import org.elasticsearch.xpack.eql.expression.function.scalar.string.BetweenFunctionProcessor;
 import org.elasticsearch.xpack.eql.expression.function.scalar.string.CIDRMatchFunctionProcessor;
@@ -20,6 +21,8 @@
 import org.elasticsearch.xpack.ql.expression.function.scalar.whitelist.InternalQlScriptUtils;
 
 import java.util.List;
+import java.util.Map;
+import java.util.function.Predicate;
 
 import static org.elasticsearch.xpack.eql.expression.predicate.operator.comparison.InsensitiveBinaryComparisonProcessor.InsensitiveBinaryComparisonOperation;
 
@@ -33,6 +36,20 @@ public class InternalEqlScriptUtils extends InternalQlScriptUtils {
     InternalEqlScriptUtils() {
     }
 
+    public static <T> Boolean multiValueDocValues(Map<String, ScriptDocValues<T>> doc, String fieldName, Predicate<T> script) {
+        ScriptDocValues<T> docValues = doc.get(fieldName);
+        if (docValues != null && docValues.isEmpty() == false) {
+            for (T value : docValues) {
+                if (script.test(value)) {
+                    return true;
+                }
+            }
+            return false;
+        }
+        // missing value means "null"
+        return script.test(null);
+    }
+
     public static Boolean seq(Object left, Object right) {
         return InsensitiveBinaryComparisonOperation.SEQ.apply(left, right);
     }

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

@@ -29,7 +29,6 @@
 import org.elasticsearch.xpack.ql.expression.predicate.Predicates;
 import org.elasticsearch.xpack.ql.expression.predicate.logical.Not;
 import org.elasticsearch.xpack.ql.expression.predicate.logical.Or;
-import org.elasticsearch.xpack.ql.expression.predicate.nulls.IsNotNull;
 import org.elasticsearch.xpack.ql.expression.predicate.nulls.IsNull;
 import org.elasticsearch.xpack.ql.expression.predicate.operator.comparison.BinaryComparison;
 import org.elasticsearch.xpack.ql.expression.predicate.operator.comparison.Equals;
@@ -40,12 +39,10 @@
 import org.elasticsearch.xpack.ql.optimizer.OptimizerRules.BinaryComparisonSimplification;
 import org.elasticsearch.xpack.ql.optimizer.OptimizerRules.BooleanFunctionEqualsElimination;
 import org.elasticsearch.xpack.ql.optimizer.OptimizerRules.BooleanSimplification;
-import org.elasticsearch.xpack.ql.optimizer.OptimizerRules.CombineBinaryComparisons;
 import org.elasticsearch.xpack.ql.optimizer.OptimizerRules.CombineDisjunctionsToIn;
 import org.elasticsearch.xpack.ql.optimizer.OptimizerRules.ConstantFolding;
 import org.elasticsearch.xpack.ql.optimizer.OptimizerRules.LiteralsOnTheRight;
 import org.elasticsearch.xpack.ql.optimizer.OptimizerRules.OptimizerRule;
-import org.elasticsearch.xpack.ql.optimizer.OptimizerRules.PropagateEquals;
 import org.elasticsearch.xpack.ql.optimizer.OptimizerRules.PruneLiteralsInOrderBy;
 import org.elasticsearch.xpack.ql.optimizer.OptimizerRules.PushDownAndCombineFilters;
 import org.elasticsearch.xpack.ql.optimizer.OptimizerRules.ReplaceSurrogateFunction;
@@ -66,7 +63,6 @@
 
 import static java.util.Arrays.asList;
 import static java.util.stream.Collectors.toList;
-import static org.elasticsearch.xpack.ql.optimizer.OptimizerRules.PropagateNullable;
 
 public class Optimizer extends RuleExecutor<LogicalPlan> {
 
@@ -85,14 +81,10 @@ protected Iterable<RuleExecutor<LogicalPlan>.Batch> batches() {
         Batch operators = new Batch("Operator Optimization",
                 new ConstantFolding(),
                 // boolean
-                new BooleanSimplification(),
+                new EqlBooleanSimplification(),
                 new LiteralsOnTheRight(),
                 new BinaryComparisonSimplification(),
                 new BooleanFunctionEqualsElimination(),
-                // needs to occur before BinaryComparison combinations
-                new PropagateEquals(),
-                new PropagateNullable(),
-                new CombineBinaryComparisons(),
                 new CombineDisjunctionsToIn(),
                 new SimplifyComparisonsArithmetics(DataTypes::areCompatible),
                 // prune/elimination
@@ -179,10 +171,9 @@ protected LogicalPlan rule(Filter filter) {
                         comparableToNull = cmp.right();
                     }
                     if (comparableToNull != null) {
-                        if (cmp instanceof Equals) {
-                            result = new IsNull(cmp.source(), comparableToNull);
-                        } else {
-                            result = new IsNotNull(cmp.source(), comparableToNull);
+                        result = new IsNull(cmp.source(), comparableToNull);
+                        if (cmp instanceof Equals == false) {
+                            result = new Not(cmp.source(), result);
                         }
                     }
                 }
@@ -200,6 +191,19 @@ protected Expression regexToEquals(RegexMatch<?> regexMatch, Literal literal) {
         }
     }
 
+    private static class EqlBooleanSimplification extends BooleanSimplification {
+
+        EqlBooleanSimplification() {
+            super();
+        }
+
+        @Override
+        protected Expression maybeSimplifyNegatable(Expression e) {
+            return null;
+        }
+
+    }
+
     static class PruneFilters extends org.elasticsearch.xpack.ql.optimizer.OptimizerRules.PruneFilters {
 
         @Override