QL: case sensitive support in EQL (#56404) (#56597)

* QL: case sensitive support in EQL (#56404)
* adds a generic startsWith function to QL
* modifies the existent EQL startsWith function to be case sensitive
aware
* improves the existent EQL startsWith function to use a prefix query
when the function is used in a case sensitive context. Same improvement
is used in SQL's newly added STARTS_WITH function.
* adds case sensitivity to EQL configuration through a case_sensitive
parameter in the eql request, as established in #54411.
The case_sensitive parameter can be specified when running queries
(default is case insensitive)

(cherry picked from commit ee5a09ea840167566e34c28c8225dc38bc6a7ae8)

docs/reference/sql/functions/string.asciidoc

@@ -422,6 +422,36 @@ SPACE(count) <1>
 include-tagged::{sql-specs}/docs/docs.csv-spec[stringSpace]
 --------------------------------------------------
 
+[[sql-functions-string-startswith]]
+==== `STARTS_WITH`
+
+.Synopsis:
+[source, sql]
+--------------------------------------------------
+STARTS_WITH(
+    source,   <1>
+    pattern)  <2>
+--------------------------------------------------
+*Input*:
+
+<1> string expression
+<2> string expression
+
+*Output*: boolean value
+
+*Description*: Returns `true` if the source expression starts with the specified pattern, `false` otherwise. The matching is case sensitive.
+If either parameters is `null`, the function returns `null`.
+
+[source, sql]
+--------------------------------------------------
+include-tagged::{sql-specs}/docs/docs.csv-spec[stringStartsWithTrue]
+--------------------------------------------------
+
+[source, sql]
+--------------------------------------------------
+include-tagged::{sql-specs}/docs/docs.csv-spec[stringStartsWithFalse]
+--------------------------------------------------
+
 [[sql-functions-string-substring]]
 ==== `SUBSTRING`
 

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

@@ -48,6 +48,7 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
     private int fetchSize = FETCH_SIZE;
     private SearchAfterBuilder searchAfterBuilder;
     private String query;
+    private boolean isCaseSensitive = false;
 
     static final String KEY_FILTER = "filter";
     static final String KEY_TIMESTAMP_FIELD = "timestamp_field";
@@ -56,6 +57,7 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
     static final String KEY_SIZE = "size";
     static final String KEY_SEARCH_AFTER = "search_after";
     static final String KEY_QUERY = "query";
+    static final String KEY_CASE_SENSITIVE = "case_sensitive";
 
     static final ParseField FILTER = new ParseField(KEY_FILTER);
     static final ParseField TIMESTAMP_FIELD = new ParseField(KEY_TIMESTAMP_FIELD);
@@ -64,6 +66,7 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
     static final ParseField SIZE = new ParseField(KEY_SIZE);
     static final ParseField SEARCH_AFTER = new ParseField(KEY_SEARCH_AFTER);
     static final ParseField QUERY = new ParseField(KEY_QUERY);
+    static final ParseField CASE_SENSITIVE = new ParseField(KEY_CASE_SENSITIVE);
 
     private static final ObjectParser<EqlSearchRequest, Void> PARSER = objectParser(EqlSearchRequest::new);
 
@@ -82,6 +85,7 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
         fetchSize = in.readVInt();
         searchAfterBuilder = in.readOptionalWriteable(SearchAfterBuilder::new);
         query = in.readString();
+        isCaseSensitive = in.readBoolean();
     }
 
     @Override
@@ -143,6 +147,7 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
         }
 
         builder.field(KEY_QUERY, query);
+        builder.field(KEY_CASE_SENSITIVE, isCaseSensitive);
 
         return builder;
     }
@@ -162,6 +167,7 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
         parser.declareField(EqlSearchRequest::setSearchAfter, SearchAfterBuilder::fromXContent, SEARCH_AFTER,
             ObjectParser.ValueType.OBJECT_ARRAY);
         parser.declareString(EqlSearchRequest::query, QUERY);
+        parser.declareBoolean(EqlSearchRequest::isCaseSensitive, CASE_SENSITIVE);
         return parser;
     }
 
@@ -230,6 +236,13 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
         return this;
     }
 
+    public boolean isCaseSensitive() { return this.isCaseSensitive; }
+
+    public EqlSearchRequest isCaseSensitive(boolean isCaseSensitive) {
+        this.isCaseSensitive = isCaseSensitive;
+        return this;
+    }
+
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
@@ -242,6 +255,7 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
         out.writeVInt(fetchSize);
         out.writeOptionalWriteable(searchAfterBuilder);
         out.writeString(query);
+        out.writeBoolean(isCaseSensitive);
     }
 
     @Override
@@ -261,7 +275,8 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
                 Objects.equals(eventCategoryField, that.eventCategoryField) &&
                 Objects.equals(implicitJoinKeyField, that.implicitJoinKeyField) &&
                 Objects.equals(searchAfterBuilder, that.searchAfterBuilder) &&
-                Objects.equals(query, that.query);
+                Objects.equals(query, that.query) &&
+                Objects.equals(isCaseSensitive, that.isCaseSensitive);
     }
 
     @Override
@@ -274,7 +289,8 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
             timestampField, eventCategoryField,
             implicitJoinKeyField,
             searchAfterBuilder,
-            query);
+            query,
+            isCaseSensitive);
     }
 
     @Override

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

@@ -55,4 +55,8 @@ public class EqlSearchRequestBuilder extends ActionRequestBuilder<EqlSearchReque
         return this;
     }
 
+    public EqlSearchRequestBuilder query(boolean isCaseSensitive) {
+        request.isCaseSensitive(isCaseSensitive);
+        return this;
+    }
 }

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

@@ -16,6 +16,7 @@ import org.elasticsearch.xpack.ql.expression.function.FunctionRegistry;
 import org.elasticsearch.xpack.ql.expression.function.UnresolvedFunction;
 import org.elasticsearch.xpack.ql.plan.logical.LogicalPlan;
 import org.elasticsearch.xpack.ql.rule.RuleExecutor;
+import org.elasticsearch.xpack.ql.session.Configuration;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -26,10 +27,12 @@ import static org.elasticsearch.xpack.eql.analysis.AnalysisUtils.resolveAgainstL
 
 public class Analyzer extends RuleExecutor<LogicalPlan> {
 
+    private final Configuration configuration;
     private final FunctionRegistry functionRegistry;
     private final Verifier verifier;
 
-    public Analyzer(FunctionRegistry functionRegistry, Verifier verifier) {
+    public Analyzer(Configuration configuration, FunctionRegistry functionRegistry, Verifier verifier) {
+        this.configuration = configuration;
         this.functionRegistry = functionRegistry;
         this.verifier = verifier;
     }
@@ -113,7 +116,7 @@ public class Analyzer extends RuleExecutor<LogicalPlan> {
                         return uf.missing(functionName, functionRegistry.listFunctions());
                     }
                     FunctionDefinition def = functionRegistry.resolveFunction(functionName);
-                    Function f = uf.buildResolved(null, def);
+                    Function f = uf.buildResolved(configuration, def);
                     return f;
                 }
                 return e;

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

@@ -9,14 +9,13 @@ package org.elasticsearch.xpack.eql.execution;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
-import org.elasticsearch.xpack.eql.analysis.Analyzer;
 import org.elasticsearch.xpack.eql.analysis.PreAnalyzer;
 import org.elasticsearch.xpack.eql.analysis.Verifier;
 import org.elasticsearch.xpack.eql.expression.function.EqlFunctionRegistry;
 import org.elasticsearch.xpack.eql.optimizer.Optimizer;
 import org.elasticsearch.xpack.eql.parser.ParserParams;
 import org.elasticsearch.xpack.eql.planner.Planner;
-import org.elasticsearch.xpack.eql.session.Configuration;
+import org.elasticsearch.xpack.eql.session.EqlConfiguration;
 import org.elasticsearch.xpack.eql.session.EqlSession;
 import org.elasticsearch.xpack.eql.session.Results;
 import org.elasticsearch.xpack.eql.stats.Metrics;
@@ -33,7 +32,7 @@ public class PlanExecutor {
     private final FunctionRegistry functionRegistry;
 
     private final PreAnalyzer preAnalyzer;
-    private final Analyzer analyzer;
+    private final Verifier verifier;
     private final Optimizer optimizer;
     private final Planner planner;
 
@@ -50,16 +49,16 @@ public class PlanExecutor {
         this.metrics = new Metrics();
 
         this.preAnalyzer = new PreAnalyzer();
-        this.analyzer = new Analyzer(functionRegistry, new Verifier());
+        this.verifier = new Verifier();
         this.optimizer = new Optimizer();
         this.planner = new Planner();
     }
 
-    private EqlSession newSession(Configuration cfg) {
+    private EqlSession newSession(EqlConfiguration cfg) {
-        return new EqlSession(client, cfg, indexResolver, preAnalyzer, analyzer, optimizer, planner, this);
+        return new EqlSession(client, cfg, indexResolver, preAnalyzer, functionRegistry, verifier, optimizer, planner, this);
     }
 
-    public void eql(Configuration cfg, String eql, ParserParams parserParams, ActionListener<Results> listener) {
+    public void eql(EqlConfiguration cfg, String eql, ParserParams parserParams, ActionListener<Results> listener) {
         newSession(cfg).eql(eql, parserParams, wrap(listener::onResponse, listener::onFailure));
     }
 

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

@@ -19,7 +19,7 @@ import org.elasticsearch.search.aggregations.Aggregation;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.tasks.TaskCancelledException;
 import org.elasticsearch.xpack.eql.querydsl.container.QueryContainer;
-import org.elasticsearch.xpack.eql.session.Configuration;
+import org.elasticsearch.xpack.eql.session.EqlConfiguration;
 import org.elasticsearch.xpack.eql.session.EqlSession;
 import org.elasticsearch.xpack.eql.session.Results;
 import org.elasticsearch.xpack.ql.expression.Attribute;
@@ -33,7 +33,7 @@ public class Querier {
 
     private static final Logger log = LogManager.getLogger(Querier.class);
 
-    private final Configuration cfg;
+    private final EqlConfiguration cfg;
     private final Client client;
     private final TimeValue keepAlive;
     private final QueryBuilder filter;

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

@@ -20,7 +20,7 @@ import org.elasticsearch.xpack.eql.execution.search.extractor.FieldHitExtractor;
 import org.elasticsearch.xpack.eql.querydsl.container.ComputedRef;
 import org.elasticsearch.xpack.eql.querydsl.container.QueryContainer;
 import org.elasticsearch.xpack.eql.querydsl.container.SearchHitFieldRef;
-import org.elasticsearch.xpack.eql.session.Configuration;
+import org.elasticsearch.xpack.eql.session.EqlConfiguration;
 import org.elasticsearch.xpack.eql.session.Results;
 import org.elasticsearch.xpack.ql.execution.search.FieldExtraction;
 import org.elasticsearch.xpack.ql.execution.search.extractor.ComputingExtractor;
@@ -43,12 +43,12 @@ class SearchAfterListener implements ActionListener<SearchResponse> {
     private final ActionListener<Results> listener;
 
     private final Client client;
-    private final Configuration cfg;
+    private final EqlConfiguration cfg;
     private final List<Attribute> output;
     private final QueryContainer container;
     private final SearchRequest request;
 
-    SearchAfterListener(ActionListener<Results> listener, Client client, Configuration cfg, List<Attribute> output,
+    SearchAfterListener(ActionListener<Results> listener, Client client, EqlConfiguration cfg, List<Attribute> output,
                                QueryContainer container, SearchRequest request) {
 
         this.listener = listener;

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/expression/function/EqlFunctionRegistry.java

@@ -6,8 +6,8 @@
 
 package org.elasticsearch.xpack.eql.expression.function;
 
-import org.elasticsearch.xpack.eql.expression.function.scalar.string.CIDRMatch;
 import org.elasticsearch.xpack.eql.expression.function.scalar.string.Between;
+import org.elasticsearch.xpack.eql.expression.function.scalar.string.CIDRMatch;
 import org.elasticsearch.xpack.eql.expression.function.scalar.string.Concat;
 import org.elasticsearch.xpack.eql.expression.function.scalar.string.EndsWith;
 import org.elasticsearch.xpack.eql.expression.function.scalar.string.IndexOf;
@@ -50,15 +50,15 @@ public class EqlFunctionRegistry extends FunctionRegistry {
                 def(ToString.class, ToString::new, "string"),
                 def(StringContains.class, StringContains::new, "stringcontains"),
                 def(Substring.class, Substring::new, "substring"),
-                def(Wildcard.class, Wildcard::new, "wildcard"),
+                def(Wildcard.class, Wildcard::new, "wildcard")
             },
         // Arithmetic
             new FunctionDefinition[] {
-                    def(Add.class, Add::new, "add"),
+                def(Add.class, Add::new, "add"),
-                    def(Div.class, Div::new, "divide"),
+                def(Div.class, Div::new, "divide"),
-                    def(Mod.class, Mod::new, "modulo"),
+                def(Mod.class, Mod::new, "modulo"),
-                    def(Mul.class, Mul::new, "multiply"),
+                def(Mul.class, Mul::new, "multiply"),
-                    def(Sub.class, Sub::new, "subtract"),
+                def(Sub.class, Sub::new, "subtract")
             }
         };
     }

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

@@ -6,115 +6,20 @@
 
 package org.elasticsearch.xpack.eql.expression.function.scalar.string;
 
+import org.elasticsearch.xpack.eql.session.EqlConfiguration;
 import org.elasticsearch.xpack.ql.expression.Expression;
-import org.elasticsearch.xpack.ql.expression.Expressions;
+import org.elasticsearch.xpack.ql.session.Configuration;
-import org.elasticsearch.xpack.ql.expression.Expressions.ParamOrdinal;
-import org.elasticsearch.xpack.ql.expression.FieldAttribute;
-import org.elasticsearch.xpack.ql.expression.function.scalar.ScalarFunction;
-import org.elasticsearch.xpack.ql.expression.gen.pipeline.Pipe;
-import org.elasticsearch.xpack.ql.expression.gen.script.ScriptTemplate;
-import org.elasticsearch.xpack.ql.expression.gen.script.Scripts;
-import org.elasticsearch.xpack.ql.tree.NodeInfo;
 import org.elasticsearch.xpack.ql.tree.Source;
-import org.elasticsearch.xpack.ql.type.DataType;
-import org.elasticsearch.xpack.ql.type.DataTypes;
 
-import java.util.Arrays;
+public class StartsWith extends org.elasticsearch.xpack.ql.expression.function.scalar.string.StartsWith {
-import java.util.List;
-import java.util.Locale;
 
-import static java.lang.String.format;
+    public StartsWith(Source source, Expression field, Expression pattern, Configuration configuration) {
-import static org.elasticsearch.xpack.eql.expression.function.scalar.string.StartsWithFunctionProcessor.doProcess;
+        super(source, field, pattern, configuration);
-import static org.elasticsearch.xpack.ql.expression.TypeResolutions.isStringAndExact;
-import static org.elasticsearch.xpack.ql.expression.gen.script.ParamsBuilder.paramsBuilder;
-
-/**
- * Function that checks if first parameter starts with the second parameter. Both parameters should be strings
- * and the function returns a boolean value. The function is case insensitive.
- */
-public class StartsWith extends ScalarFunction {
-
-    private final Expression source;
-    private final Expression pattern;
-
-    public StartsWith(Source source, Expression src, Expression pattern) {
-        super(source, Arrays.asList(src, pattern));
-        this.source = src;
-        this.pattern = pattern;
     }
 
     @Override
-    protected TypeResolution resolveType() {
+    public boolean isCaseSensitive() {
-        if (!childrenResolved()) {
+        return ((EqlConfiguration) configuration()).isCaseSensitive();
-            return new TypeResolution("Unresolved children");
-        }
-
-        TypeResolution sourceResolution = isStringAndExact(source, sourceText(), ParamOrdinal.FIRST);
-        if (sourceResolution.unresolved()) {
-            return sourceResolution;
-        }
-
-        return isStringAndExact(pattern, sourceText(), ParamOrdinal.SECOND);
     }
 
-    @Override
+}
-    protected Pipe makePipe() {
-        return new StartsWithFunctionPipe(source(), this, Expressions.pipe(source), Expressions.pipe(pattern));
-    }
-
-    @Override
-    public boolean foldable() {
-        return source.foldable() && pattern.foldable();
-    }
-
-    @Override
-    public Object fold() {
-        return doProcess(source.fold(), pattern.fold());
-    }
-
-    @Override
-    protected NodeInfo<? extends Expression> info() {
-        return NodeInfo.create(this, StartsWith::new, source, pattern);
-    }
-
-    @Override
-    public ScriptTemplate asScript() {
-        ScriptTemplate sourceScript = asScript(source);
-        ScriptTemplate patternScript = asScript(pattern);
-
-        return asScriptFrom(sourceScript, patternScript);
-    }
-    
-    protected ScriptTemplate asScriptFrom(ScriptTemplate sourceScript, ScriptTemplate patternScript) {
-        return new ScriptTemplate(format(Locale.ROOT, formatTemplate("{eql}.%s(%s,%s)"),
-                "startsWith",
-                sourceScript.template(),
-                patternScript.template()),
-                paramsBuilder()
-                    .script(sourceScript.params())
-                    .script(patternScript.params())
-                    .build(), dataType());
-    }
-
-    @Override
-    public ScriptTemplate scriptWithField(FieldAttribute field) {
-        return new ScriptTemplate(processScript(Scripts.DOC_VALUE),
-                paramsBuilder().variable(field.exactAttribute().name()).build(),
-                dataType());
-    }
-
-    @Override
-    public DataType dataType() {
-        return DataTypes.BOOLEAN;
-    }
-
-    @Override
-    public Expression replaceChildren(List<Expression> newChildren) {
-        if (newChildren.size() != 2) {
-            throw new IllegalArgumentException("expected [2] children but received [" + newChildren.size() + "]");
-        }
-
-        return new StartsWith(source(), newChildren.get(0), newChildren.get(1));
-    }
-
-}

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

@@ -11,7 +11,6 @@ import org.elasticsearch.xpack.eql.expression.function.scalar.string.ConcatFunct
 import org.elasticsearch.xpack.eql.expression.function.scalar.string.EndsWithFunctionProcessor;
 import org.elasticsearch.xpack.eql.expression.function.scalar.string.IndexOfFunctionProcessor;
 import org.elasticsearch.xpack.eql.expression.function.scalar.string.LengthFunctionProcessor;
-import org.elasticsearch.xpack.eql.expression.function.scalar.string.StartsWithFunctionProcessor;
 import org.elasticsearch.xpack.eql.expression.function.scalar.string.StringContainsFunctionProcessor;
 import org.elasticsearch.xpack.eql.expression.function.scalar.string.SubstringFunctionProcessor;
 import org.elasticsearch.xpack.eql.expression.function.scalar.string.ToStringFunctionProcessor;
@@ -48,10 +47,6 @@ public class InternalEqlScriptUtils extends InternalQlScriptUtils {
         return (Integer) LengthFunctionProcessor.doProcess(s);
     }
 
-    public static Boolean startsWith(String s, String pattern) {
-        return (Boolean) StartsWithFunctionProcessor.doProcess(s, pattern);
-    }
-
     public static String string(Object s) {
         return (String) ToStringFunctionProcessor.doProcess(s);
     }

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

@@ -26,7 +26,7 @@ import org.elasticsearch.xpack.eql.action.EqlSearchResponse;
 import org.elasticsearch.xpack.eql.action.EqlSearchTask;
 import org.elasticsearch.xpack.eql.execution.PlanExecutor;
 import org.elasticsearch.xpack.eql.parser.ParserParams;
-import org.elasticsearch.xpack.eql.session.Configuration;
+import org.elasticsearch.xpack.eql.session.EqlConfiguration;
 import org.elasticsearch.xpack.eql.session.Results;
 
 import java.time.ZoneId;
@@ -69,8 +69,8 @@ public class TransportEqlSearchAction extends HandledTransportAction<EqlSearchRe
             .fieldTimestamp(request.timestampField())
             .implicitJoinKey(request.implicitJoinKeyField());
 
-        Configuration cfg = new Configuration(request.indices(), zoneId, username, clusterName, filter, timeout, request.fetchSize(),
+        EqlConfiguration cfg = new EqlConfiguration(request.indices(), zoneId, username, clusterName, filter, timeout, request.fetchSize(),
-                includeFrozen, clientId, new TaskId(nodeId, task.getId()), task::isCancelled);
+                includeFrozen, request.isCaseSensitive(), clientId, new TaskId(nodeId, task.getId()), task::isCancelled);
         planExecutor.eql(cfg, request.query(), params, wrap(r -> listener.onResponse(createResponse(r)), listener::onFailure));
     }
 

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

@@ -14,7 +14,7 @@ import org.elasticsearch.tasks.TaskId;
 import java.time.ZoneId;
 import java.util.function.Supplier;
 
-public class Configuration extends org.elasticsearch.xpack.ql.session.Configuration {
+public class EqlConfiguration extends org.elasticsearch.xpack.ql.session.Configuration {
 
     private final String[] indices;
     private final TimeValue requestTimeout;
@@ -23,12 +23,14 @@ public class Configuration extends org.elasticsearch.xpack.ql.session.Configurat
     private final boolean includeFrozenIndices;
     private final Supplier<Boolean> isCancelled;
     private final TaskId taskId;
+    private final boolean isCaseSensitive;
 
     @Nullable
     private final QueryBuilder filter;
 
-    public Configuration(String[] indices, ZoneId zi, String username, String clusterName, QueryBuilder filter, TimeValue requestTimeout,
+    public EqlConfiguration(String[] indices, ZoneId zi, String username, String clusterName, QueryBuilder filter, TimeValue requestTimeout,
-                         int size, boolean includeFrozen, String clientId, TaskId taskId, Supplier<Boolean> isCancelled) {
+                         int size, boolean includeFrozen, boolean isCaseSensitive, String clientId, TaskId taskId,
+                         Supplier<Boolean> isCancelled) {
 
         super(zi, username, clusterName);
 
@@ -38,6 +40,7 @@ public class Configuration extends org.elasticsearch.xpack.ql.session.Configurat
         this.size = size;
         this.clientId = clientId;
         this.includeFrozenIndices = includeFrozen;
+        this.isCaseSensitive = isCaseSensitive;
         this.taskId = taskId;
         this.isCancelled = isCancelled;
     }
@@ -66,6 +69,10 @@ public class Configuration extends org.elasticsearch.xpack.ql.session.Configurat
         return includeFrozenIndices;
     }
 
+    public boolean isCaseSensitive() {
+        return isCaseSensitive;
+    }
+
     public boolean isCancelled() {
         return isCancelled.get();
     }

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

@@ -13,12 +13,14 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.tasks.TaskCancelledException;
 import org.elasticsearch.xpack.eql.analysis.Analyzer;
 import org.elasticsearch.xpack.eql.analysis.PreAnalyzer;
+import org.elasticsearch.xpack.eql.analysis.Verifier;
 import org.elasticsearch.xpack.eql.execution.PlanExecutor;
 import org.elasticsearch.xpack.eql.optimizer.Optimizer;
 import org.elasticsearch.xpack.eql.parser.EqlParser;
 import org.elasticsearch.xpack.eql.parser.ParserParams;
 import org.elasticsearch.xpack.eql.plan.physical.PhysicalPlan;
 import org.elasticsearch.xpack.eql.planner.Planner;
+import org.elasticsearch.xpack.ql.expression.function.FunctionRegistry;
 import org.elasticsearch.xpack.ql.index.IndexResolver;
 import org.elasticsearch.xpack.ql.plan.logical.LogicalPlan;
 
@@ -27,7 +29,7 @@ import static org.elasticsearch.action.ActionListener.wrap;
 public class EqlSession {
 
     private final Client client;
-    private final Configuration configuration;
+    private final EqlConfiguration configuration;
     private final IndexResolver indexResolver;
 
     private final PreAnalyzer preAnalyzer;
@@ -35,14 +37,14 @@ public class EqlSession {
     private final Optimizer optimizer;
     private final Planner planner;
 
-    public EqlSession(Client client, Configuration cfg, IndexResolver indexResolver, PreAnalyzer preAnalyzer, Analyzer analyzer,
+    public EqlSession(Client client, EqlConfiguration cfg, IndexResolver indexResolver, PreAnalyzer preAnalyzer,
-            Optimizer optimizer, Planner planner, PlanExecutor planExecutor) {
+            FunctionRegistry functionRegistry, Verifier verifier, Optimizer optimizer, Planner planner, PlanExecutor planExecutor) {
 
         this.client = new ParentTaskAssigningClient(client, cfg.getTaskId());
         this.configuration = cfg;
         this.indexResolver = indexResolver;
         this.preAnalyzer = preAnalyzer;
-        this.analyzer = analyzer;
+        this.analyzer = new Analyzer(cfg, functionRegistry, verifier);
         this.optimizer = optimizer;
         this.planner = planner;
     }
@@ -55,7 +57,7 @@ public class EqlSession {
         return optimizer;
     }
 
-    public Configuration configuration() {
+    public EqlConfiguration configuration() {
         return configuration;
     }
 

x-pack/plugin/eql/src/main/resources/org/elasticsearch/xpack/eql/plugin/eql_whitelist.txt

@@ -18,6 +18,11 @@ class org.elasticsearch.xpack.ql.expression.function.scalar.whitelist.InternalQl
   double nullSafeSortNumeric(Number)
   String nullSafeSortString(Object)
 
+#
+# ASCII Functions
+#
+  Boolean startsWith(String, String, Boolean)
+
 #
 # Comparison
 #
@@ -65,7 +70,6 @@ class org.elasticsearch.xpack.eql.expression.function.scalar.whitelist.InternalE
   Boolean endsWith(String, String)
   Integer indexOf(String, String, Number)
   Integer length(String)
-  Boolean startsWith(String, String)
   String  string(Object)
   Boolean stringContains(String, String)
   String  substring(String, Number, Number)

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

@@ -10,7 +10,7 @@ import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.xpack.eql.action.EqlSearchAction;
 import org.elasticsearch.xpack.eql.action.EqlSearchTask;
-import org.elasticsearch.xpack.eql.session.Configuration;
+import org.elasticsearch.xpack.eql.session.EqlConfiguration;
 
 import java.util.Collections;
 
@@ -26,12 +26,12 @@ public final class EqlTestUtils {
     private EqlTestUtils() {
     }
 
-    public static final Configuration TEST_CFG = new Configuration(new String[]{"none"}, org.elasticsearch.xpack.ql.util.DateUtils.UTC,
+    public static final EqlConfiguration TEST_CFG = new EqlConfiguration(new String[]{"none"},
-            "nobody", "cluster", null, TimeValue.timeValueSeconds(30), -1, false, "",
+            org.elasticsearch.xpack.ql.util.DateUtils.UTC, "nobody", "cluster", null, TimeValue.timeValueSeconds(30), -1, false, false, "",
             new TaskId(randomAlphaOfLength(10), randomNonNegativeLong()), () -> false);
 
-    public static Configuration randomConfiguration() {
+    public static EqlConfiguration randomConfiguration() {
-        return new Configuration(new String[]{randomAlphaOfLength(16)},
+        return new EqlConfiguration(new String[]{randomAlphaOfLength(16)},
             randomZone(),
             randomAlphaOfLength(16),
             randomAlphaOfLength(16),
@@ -39,6 +39,22 @@ public final class EqlTestUtils {
             new TimeValue(randomNonNegativeLong()),
             randomIntBetween(5, 100),
             randomBoolean(),
+            randomBoolean(),
+            randomAlphaOfLength(16),
+            new TaskId(randomAlphaOfLength(10), randomNonNegativeLong()),
+            () -> false);
+    }
+
+    public static EqlConfiguration randomConfigurationWithCaseSensitive(boolean isCaseSensitive) {
+        return new EqlConfiguration(new String[]{randomAlphaOfLength(16)},
+            randomZone(),
+            randomAlphaOfLength(16),
+            randomAlphaOfLength(16),
+            null,
+            new TimeValue(randomNonNegativeLong()),
+            randomIntBetween(5, 100),
+            randomBoolean(),
+            isCaseSensitive,
             randomAlphaOfLength(16),
             new TaskId(randomAlphaOfLength(10), randomNonNegativeLong()),
             () -> false);

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

@@ -45,10 +45,13 @@ public class EqlRequestParserTests extends ESTestCase {
         assertParsingErrorMessage("{\"size\" : \"foo\"}", "failed to parse field [size]", EqlSearchRequest::fromXContent);
         assertParsingErrorMessage("{\"query\" : 123}", "query doesn't support values of type: VALUE_NUMBER",
             EqlSearchRequest::fromXContent);
-
         assertParsingErrorMessage("{\"query\" : \"whatever\", \"size\":\"abc\"}", "failed to parse field [size]",
             EqlSearchRequest::fromXContent);
+        assertParsingErrorMessage("{\"case_sensitive\" : \"whatever\"}", "failed to parse field [case_sensitive]",
+            EqlSearchRequest::fromXContent);
 
+        boolean setIsCaseSensitive = randomBoolean();
+        boolean isCaseSensitive = randomBoolean();
         EqlSearchRequest request = generateRequest("endgame-*", "{\"filter\" : {\"match\" : {\"foo\":\"bar\"}}, "
             + "\"timestamp_field\" : \"tsf\", "
             + "\"event_category_field\" : \"etf\","
@@ -56,6 +59,7 @@ public class EqlRequestParserTests extends ESTestCase {
             + "\"search_after\" : [ 12345678, \"device-20184\", \"/user/local/foo.exe\", \"2019-11-26T00:45:43.542\" ],"
             + "\"size\" : \"101\","
             + "\"query\" : \"file where user != 'SYSTEM' by file_path\""
+            + (setIsCaseSensitive ? (",\"case_sensitive\" : " + isCaseSensitive) : "")
             + "}", EqlSearchRequest::fromXContent);
         assertArrayEquals(new String[]{"endgame-*"}, request.indices());
         assertNotNull(request.query());
@@ -69,6 +73,7 @@ public class EqlRequestParserTests extends ESTestCase {
         assertArrayEquals(new Object[]{12345678, "device-20184", "/user/local/foo.exe", "2019-11-26T00:45:43.542"}, request.searchAfter());
         assertEquals(101, request.fetchSize());
         assertEquals("file where user != 'SYSTEM' by file_path", request.query());
+        assertEquals(setIsCaseSensitive && isCaseSensitive, request.isCaseSensitive());
     }
 
     private EqlSearchRequest generateRequest(String index, String json, Function<XContentParser, EqlSearchRequest> fromXContent)

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

@@ -6,6 +6,7 @@
 package org.elasticsearch.xpack.eql.analysis;
 
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.eql.EqlTestUtils;
 import org.elasticsearch.xpack.eql.expression.function.EqlFunctionRegistry;
 import org.elasticsearch.xpack.eql.parser.EqlParser;
 import org.elasticsearch.xpack.eql.parser.ParsingException;
@@ -35,7 +36,7 @@ public class VerifierTests extends ESTestCase {
 
     private LogicalPlan accept(IndexResolution resolution, String eql) {
         PreAnalyzer preAnalyzer = new PreAnalyzer();
-        Analyzer analyzer = new Analyzer(new EqlFunctionRegistry(), new Verifier());
+        Analyzer analyzer = new Analyzer(EqlTestUtils.TEST_CFG, new EqlFunctionRegistry(), new Verifier());
         return analyzer.analyze(preAnalyzer.preAnalyze(parser.createStatement(eql), resolution));
     }
 

x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/expression/function/scalar/string/StartsWithProcessorTests.java

@@ -6,49 +6,30 @@
 
 package org.elasticsearch.xpack.eql.expression.function.scalar.string;
 
-import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.ql.expression.Expression;
-import org.elasticsearch.xpack.ql.QlIllegalArgumentException;
+import org.elasticsearch.xpack.ql.expression.function.scalar.string.StartsWith;
-import org.elasticsearch.xpack.ql.expression.Literal;
+import org.elasticsearch.xpack.ql.session.Configuration;
-import org.elasticsearch.xpack.ql.expression.LiteralTests;
+import org.elasticsearch.xpack.ql.tree.Source;
 
-import static org.elasticsearch.xpack.ql.expression.function.scalar.FunctionTestUtils.l;
+import java.util.function.Supplier;
-import static org.elasticsearch.xpack.ql.tree.Source.EMPTY;
-import static org.elasticsearch.xpack.ql.type.DataTypes.KEYWORD;
-import static org.hamcrest.Matchers.startsWith;
 
-public class StartsWithProcessorTests extends ESTestCase {
+import static org.elasticsearch.xpack.eql.EqlTestUtils.randomConfigurationWithCaseSensitive;
 
-    public void testStartsWithFunctionWithValidInput() {
+public class StartsWithProcessorTests extends org.elasticsearch.xpack.ql.expression.function.scalar.string.StartsWithProcessorTests {
-        assertEquals(true, new StartsWith(EMPTY, l("foobarbar"), l("f")).makePipe().asProcessor().process(null));
+
-        assertEquals(false, new StartsWith(EMPTY, l("foobar"), l("bar")).makePipe().asProcessor().process(null));
+    @Override
-        assertEquals(false, new StartsWith(EMPTY, l("foo"), l("foobar")).makePipe().asProcessor().process(null));
+    protected Supplier<Boolean> isCaseSensitiveGenerator() {
-        assertEquals(true, new StartsWith(EMPTY, l("foobar"), l("")).makePipe().asProcessor().process(null));
+        return () -> randomBoolean();
-        assertEquals(true, new StartsWith(EMPTY, l("foo"), l("foo")).makePipe().asProcessor().process(null));
-        assertEquals(true, new StartsWith(EMPTY, l("foo"), l("FO")).makePipe().asProcessor().process(null));
-        assertEquals(true, new StartsWith(EMPTY, l("foo"), l("FOo")).makePipe().asProcessor().process(null));
-        assertEquals(true, new StartsWith(EMPTY, l('f'), l('f')).makePipe().asProcessor().process(null));
-        assertEquals(false, new StartsWith(EMPTY, l(""), l("bar")).makePipe().asProcessor().process(null));
-        assertEquals(null, new StartsWith(EMPTY, l(null), l("bar")).makePipe().asProcessor().process(null));
-        assertEquals(null, new StartsWith(EMPTY, l("foo"), l(null)).makePipe().asProcessor().process(null));
-        assertEquals(null, new StartsWith(EMPTY, l(null), l(null)).makePipe().asProcessor().process(null));
-    }
-    
-    public void testStartsWithFunctionInputsValidation() {
-        QlIllegalArgumentException siae = expectThrows(QlIllegalArgumentException.class,
-                () -> new StartsWith(EMPTY, l(5), l("foo")).makePipe().asProcessor().process(null));
-        assertEquals("A string/char is required; received [5]", siae.getMessage());
-        siae = expectThrows(QlIllegalArgumentException.class,
-                () -> new StartsWith(EMPTY, l("bar"), l(false)).makePipe().asProcessor().process(null));
-        assertEquals("A string/char is required; received [false]", siae.getMessage());
     }
 
-    public void testStartsWithFunctionWithRandomInvalidDataType() {
+    @Override
-        Literal literal = randomValueOtherThanMany(v -> v.dataType() == KEYWORD, () -> LiteralTests.randomLiteral());
+    protected Supplier<Configuration> configurationGenerator() {
-        QlIllegalArgumentException siae = expectThrows(QlIllegalArgumentException.class,
+        return () -> randomConfigurationWithCaseSensitive(isCaseSensitive);
-                () -> new StartsWith(EMPTY, literal, l("foo")).makePipe().asProcessor().process(null));
-        assertThat(siae.getMessage(), startsWith("A string/char is required; received"));
-        siae = expectThrows(QlIllegalArgumentException.class,
-                () -> new StartsWith(EMPTY, l("foo"), literal).makePipe().asProcessor().process(null));
-        assertThat(siae.getMessage(), startsWith("A string/char is required; received"));
     }
+
+    @Override
+    protected Supplier<StartsWith> startsWithInstantiator(Source source, Expression field, Expression pattern) {
+        return () -> new org.elasticsearch.xpack.eql.expression.function.scalar.string.StartsWith(source, field, pattern, config);
+    }
+
 }

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

@@ -30,6 +30,8 @@ import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 
+import static org.elasticsearch.xpack.eql.EqlTestUtils.TEST_CFG;
+
 public class OptimizerTests extends ESTestCase {
 
 
@@ -47,7 +49,7 @@ public class OptimizerTests extends ESTestCase {
 
     private LogicalPlan accept(IndexResolution resolution, String eql) {
         PreAnalyzer preAnalyzer = new PreAnalyzer();
-        Analyzer analyzer = new Analyzer(new EqlFunctionRegistry(), new Verifier());
+        Analyzer analyzer = new Analyzer(TEST_CFG, new EqlFunctionRegistry(), new Verifier());
         Optimizer optimizer = new Optimizer();
         return optimizer.optimize(analyzer.analyze(preAnalyzer.preAnalyze(parser.createStatement(eql), resolution)));
     }

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

@@ -7,6 +7,7 @@
 package org.elasticsearch.xpack.eql.planner;
 
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.eql.EqlTestUtils;
 import org.elasticsearch.xpack.eql.analysis.Analyzer;
 import org.elasticsearch.xpack.eql.analysis.PreAnalyzer;
 import org.elasticsearch.xpack.eql.analysis.Verifier;
@@ -14,6 +15,7 @@ import org.elasticsearch.xpack.eql.expression.function.EqlFunctionRegistry;
 import org.elasticsearch.xpack.eql.optimizer.Optimizer;
 import org.elasticsearch.xpack.eql.parser.EqlParser;
 import org.elasticsearch.xpack.eql.plan.physical.PhysicalPlan;
+import org.elasticsearch.xpack.eql.session.EqlConfiguration;
 import org.elasticsearch.xpack.ql.index.EsIndex;
 import org.elasticsearch.xpack.ql.index.IndexResolution;
 
@@ -22,7 +24,8 @@ import static org.elasticsearch.xpack.ql.type.TypesTests.loadMapping;
 public abstract class AbstractQueryFolderTestCase extends ESTestCase {
     protected EqlParser parser = new EqlParser();
     protected PreAnalyzer preAnalyzer = new PreAnalyzer();
-    protected Analyzer analyzer = new Analyzer(new EqlFunctionRegistry(), new Verifier());
+    protected EqlConfiguration configuration = EqlTestUtils.randomConfiguration();
+    protected Analyzer analyzer = new Analyzer(configuration, new EqlFunctionRegistry(), new Verifier());
     protected Optimizer optimizer = new Optimizer();
     protected Planner planner = new Planner();
 

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

@@ -11,12 +11,14 @@ import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.xpack.eql.plan.physical.EsQueryExec;
 import org.elasticsearch.xpack.eql.plan.physical.PhysicalPlan;
+import org.junit.Assume;
 
 import java.io.BufferedReader;
 import java.io.InputStreamReader;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
+import java.util.Locale;
 import java.util.Map;
 
 import static org.elasticsearch.xpack.ql.type.DataTypes.KEYWORD;
@@ -68,18 +70,14 @@ public class QueryFolderOkTests extends AbstractQueryFolderTestCase {
                         throw new IllegalArgumentException("Duplicate test name '" + line + "' at line " + lineNumber
                                 + " (previously seen at line " + previousName + ")");
                     }
-                }
+                } else if (query == null) {
-
-                else if (query == null) {
                     sb.append(line);
                     if (line.endsWith(";")) {
                         sb.setLength(sb.length() - 1);
                         query = sb.toString();
                         sb.setLength(0);
                     }
-                }
+                } else {
-
-                else {
                     boolean done = false;
                     if (line.endsWith(";")) {
                         line = line.substring(0, line.length() - 1);
@@ -89,7 +87,6 @@ public class QueryFolderOkTests extends AbstractQueryFolderTestCase {
                     if (line.equals("null") == false) {
                         expectations.add(line);
                     }
-
                     if (done) {
                         // Add and zero out for the next spec
                         addSpec(arr, name, query, expectations.isEmpty() ? null : expectations.toArray());
@@ -114,6 +111,11 @@ public class QueryFolderOkTests extends AbstractQueryFolderTestCase {
     }
 
     public void test() {
+        // skip tests that do not make sense from case sensitivity point of view
+        boolean isCaseSensitiveValidTest = name.toLowerCase(Locale.ROOT).endsWith("-casesensitive") && configuration.isCaseSensitive() 
+            || name.toLowerCase(Locale.ROOT).endsWith("-caseinsensitive") && configuration.isCaseSensitive() == false;
+        Assume.assumeTrue(isCaseSensitiveValidTest);
+
         PhysicalPlan p = plan(query);
         assertEquals(EsQueryExec.class, p.getClass());
         EsQueryExec eqe = (EsQueryExec) p;

x-pack/plugin/eql/src/test/resources/queryfolder_tests.txt

@@ -146,12 +146,27 @@ InternalEqlScriptUtils.length(InternalQlScriptUtils.docValue(doc,params.v0)),par
 "params":{"v0":"constant_keyword","v1":5}
 ;
 
-startsWithFunction
+startsWithFunction-caseInSensitive
 process where startsWith(user_name, 'A')
 ;
-"script":{"source":"InternalQlScriptUtils.nullSafeFilter(InternalEqlScriptUtils.startsWith(
+"script":{"source":"InternalQlScriptUtils.nullSafeFilter(InternalQlScriptUtils.startsWith(
-InternalQlScriptUtils.docValue(doc,params.v0),params.v1))",
+InternalQlScriptUtils.docValue(doc,params.v0),params.v1,params.v2))",
-"params":{"v0":"user_name","v1":"A"}
+"params":{"v0":"user_name","v1":"A","v2":false}}
+;
+
+startsWithFunctionSimple-caseSensitive
+process where startsWith(user_name, 'A')
+;
+{"bool":{"must":[{"term":{"event.category":{"value":"process","boost":1.0}}},
+{"prefix":{"user_name":{"value":"A","boost":1.0}}}],"adjust_pure_negative":true,"boost":1.0}}
+;
+
+startsWithFunctionWithCondition-caseSensitive
+process where startsWith(user_name, 'A') or startsWith(user_name, 'B')
+;
+{"bool":{"must":[{"term":{"event.category":{"value":"process","boost":1.0}}},
+{"bool":{"should":[{"prefix":{"user_name":{"value":"A","boost":1.0}}},
+{"prefix":{"user_name":{"value":"B","boost":1.0}}}],"adjust_pure_negative":true,"boost":1.0}}]
 ;
 
 stringContains

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/expression/function/FunctionRegistry.java

@@ -368,7 +368,7 @@ public class FunctionRegistry {
         return new FunctionDefinition(primaryName, unmodifiableList(aliases), function, datetime, realBuilder);
     }
 
-    protected interface FunctionBuilder {
+    public interface FunctionBuilder {
         Function build(Source source, List<Expression> children, boolean distinct, Configuration cfg);
     }
 
@@ -483,4 +483,26 @@ public class FunctionRegistry {
     protected interface TwoParametersVariadicBuilder<T> {
         T build(Source source, Expression src, List<Expression> remaining);
     }
+
+    /**
+     * Build a {@linkplain FunctionDefinition} for a binary function that is case sensitive aware.
+     */
+    @SuppressWarnings("overloads")  // These are ambiguous if you aren't using ctor references but we always do
+    public static <T extends Function> FunctionDefinition def(Class<T> function,
+        ScalarBiFunctionConfigurationAwareBuilder<T> ctorRef, String... names) {
+        FunctionBuilder builder = (source, children, distinct, cfg) -> {
+            if (children.size() != 2) {
+                throw new QlIllegalArgumentException("expects exactly two arguments");
+            }
+            if (distinct) {
+                throw new QlIllegalArgumentException("does not support DISTINCT yet it was specified");
+            }
+            return ctorRef.build(source, children.get(0), children.get(1), cfg);
+        };
+        return def(function, builder, true, names);
+    }
+
+    protected interface ScalarBiFunctionConfigurationAwareBuilder<T> {
+        T build(Source source, Expression e1, Expression e2, Configuration configuration);
+    }
 }

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/expression/function/scalar/ConfigurationFunction.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.ql.expression.function.scalar;
+
+import org.elasticsearch.xpack.ql.expression.Expression;
+import org.elasticsearch.xpack.ql.session.Configuration;
+import org.elasticsearch.xpack.ql.tree.Source;
+
+import java.util.List;
+
+public abstract class ConfigurationFunction extends ScalarFunction {
+
+    private final Configuration configuration;
+
+    protected ConfigurationFunction(Source source, List<Expression> fields, Configuration configuration) {
+        super(source, fields);
+        this.configuration = configuration;
+    }
+
+    public Configuration configuration() {
+        return configuration;
+    }
+}

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/expression/function/scalar/string/CaseSensitiveScalarFunction.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.ql.expression.function.scalar.string;
+
+import org.elasticsearch.xpack.ql.expression.Expression;
+import org.elasticsearch.xpack.ql.expression.function.scalar.ConfigurationFunction;
+import org.elasticsearch.xpack.ql.session.Configuration;
+import org.elasticsearch.xpack.ql.tree.Source;
+
+import java.util.List;
+import java.util.Objects;
+
+public abstract class CaseSensitiveScalarFunction extends ConfigurationFunction {
+
+    protected CaseSensitiveScalarFunction(Source source, List<Expression> fields, Configuration configuration) {
+        super(source, fields, configuration);
+    }
+
+    public abstract boolean isCaseSensitive();
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), isCaseSensitive());
+    }
+
+    @Override
+    public boolean equals(Object other) {
+        return super.equals(other) && Objects.equals(((CaseSensitiveScalarFunction) other).isCaseSensitive(), isCaseSensitive());
+    }
+}

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/expression/function/scalar/string/StartsWith.java

@@ -0,0 +1,132 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.ql.expression.function.scalar.string;
+
+import org.elasticsearch.xpack.ql.expression.Expression;
+import org.elasticsearch.xpack.ql.expression.Expressions;
+import org.elasticsearch.xpack.ql.expression.Expressions.ParamOrdinal;
+import org.elasticsearch.xpack.ql.expression.FieldAttribute;
+import org.elasticsearch.xpack.ql.expression.gen.pipeline.Pipe;
+import org.elasticsearch.xpack.ql.expression.gen.script.ParamsBuilder;
+import org.elasticsearch.xpack.ql.expression.gen.script.ScriptTemplate;
+import org.elasticsearch.xpack.ql.expression.gen.script.Scripts;
+import org.elasticsearch.xpack.ql.session.Configuration;
+import org.elasticsearch.xpack.ql.tree.NodeInfo;
+import org.elasticsearch.xpack.ql.tree.Source;
+import org.elasticsearch.xpack.ql.type.DataType;
+import org.elasticsearch.xpack.ql.type.DataTypes;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.elasticsearch.xpack.ql.expression.TypeResolutions.isStringAndExact;
+import static org.elasticsearch.xpack.ql.expression.function.scalar.string.StartsWithFunctionProcessor.doProcess;
+import static org.elasticsearch.xpack.ql.expression.gen.script.ParamsBuilder.paramsBuilder;
+
+/**
+ * Function that checks if first parameter starts with the second parameter. Both parameters should be strings
+ * and the function returns a boolean value. The function is case insensitive.
+ */
+public class StartsWith extends CaseSensitiveScalarFunction {
+
+    private final Expression field;
+    private final Expression pattern;
+
+    public StartsWith(Source source, Expression field, Expression pattern, Configuration configuration) {
+        super(source, Arrays.asList(field, pattern), configuration);
+        this.field = field;
+        this.pattern = pattern;
+    }
+
+    @Override
+    protected TypeResolution resolveType() {
+        if (!childrenResolved()) {
+            return new TypeResolution("Unresolved children");
+        }
+
+        TypeResolution fieldResolution = isStringAndExact(field, sourceText(), ParamOrdinal.FIRST);
+        if (fieldResolution.unresolved()) {
+            return fieldResolution;
+        }
+
+        return isStringAndExact(pattern, sourceText(), ParamOrdinal.SECOND);
+    }
+
+    public Expression field() {
+        return field;
+    }
+
+    public Expression pattern() {
+        return pattern;
+    }
+
+    @Override
+    public boolean isCaseSensitive() {
+        return true;
+    }
+
+    @Override
+    public Pipe makePipe() {
+        return new StartsWithFunctionPipe(source(), this, Expressions.pipe(field), Expressions.pipe(pattern), isCaseSensitive());
+    }
+
+    @Override
+    public boolean foldable() {
+        return field.foldable() && pattern.foldable();
+    }
+
+    @Override
+    public Object fold() {
+        return doProcess(field.fold(), pattern.fold(), isCaseSensitive());
+    }
+
+    @Override
+    protected NodeInfo<? extends Expression> info() {
+        return NodeInfo.create(this, StartsWith::new, field, pattern, configuration());
+    }
+
+    @Override
+    public ScriptTemplate asScript() {
+        ScriptTemplate fieldScript = asScript(field);
+        ScriptTemplate patternScript = asScript(pattern);
+
+        return asScriptFrom(fieldScript, patternScript);
+    }
+    
+    protected ScriptTemplate asScriptFrom(ScriptTemplate fieldScript, ScriptTemplate patternScript) {
+        ParamsBuilder params = paramsBuilder();
+
+        String template = formatTemplate("{ql}.startsWith(" + fieldScript.template() + ", " + patternScript.template() + ", {})");
+        params.script(fieldScript.params())
+              .script(patternScript.params())
+              .variable(isCaseSensitive());
+        
+        return new ScriptTemplate(template, params.build(), dataType());
+    }
+
+    @Override
+    public ScriptTemplate scriptWithField(FieldAttribute field) {
+        return new ScriptTemplate(processScript(Scripts.DOC_VALUE),
+                paramsBuilder().variable(field.exactAttribute().name()).build(),
+                dataType());
+    }
+
+    @Override
+    public DataType dataType() {
+        return DataTypes.BOOLEAN;
+    }
+
+    @Override
+    public Expression replaceChildren(List<Expression> newChildren) {
+        if (newChildren.size() != 2) {
+            throw new IllegalArgumentException("expected [2] children but received [" + newChildren.size() + "]");
+        }
+
+        return new StartsWith(source(), newChildren.get(0), newChildren.get(1), configuration());
+    }
+
+}

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/expression/function/scalar/string/StartsWithFunctionPipe.java

@@ -3,7 +3,7 @@
  * or more contributor license agreements. Licensed under the Elastic License;
  * you may not use this file except in compliance with the Elastic License.
  */
-package org.elasticsearch.xpack.eql.expression.function.scalar.string;
+package org.elasticsearch.xpack.ql.expression.function.scalar.string;
 
 import org.elasticsearch.xpack.ql.execution.search.QlSourceBuilder;
 import org.elasticsearch.xpack.ql.expression.Expression;
@@ -17,13 +17,15 @@ import java.util.Objects;
 
 public class StartsWithFunctionPipe extends Pipe {
 
-    private final Pipe source;
+    private final Pipe field;
     private final Pipe pattern;
+    private final boolean isCaseSensitive;
 
-    public StartsWithFunctionPipe(Source source, Expression expression, Pipe src, Pipe pattern) {
+    public StartsWithFunctionPipe(Source source, Expression expression, Pipe field, Pipe pattern, boolean isCaseSensitive) {
-        super(source, expression, Arrays.asList(src, pattern));
+        super(source, expression, Arrays.asList(field, pattern));
-        this.source = src;
+        this.field = field;
         this.pattern = pattern;
+        this.isCaseSensitive = isCaseSensitive;
     }
 
     @Override
@@ -36,55 +38,59 @@ public class StartsWithFunctionPipe extends Pipe {
 
     @Override
     public final Pipe resolveAttributes(AttributeResolver resolver) {
-        Pipe newSource = source.resolveAttributes(resolver);
+        Pipe newField = field.resolveAttributes(resolver);
         Pipe newPattern = pattern.resolveAttributes(resolver);
-        if (newSource == source && newPattern == pattern) {
+        if (newField == field && newPattern == pattern) {
             return this;
         }
-        return replaceChildren(newSource, newPattern);
+        return replaceChildren(newField, newPattern);
     }
 
     @Override
     public boolean supportedByAggsOnlyQuery() {
-        return source.supportedByAggsOnlyQuery() && pattern.supportedByAggsOnlyQuery();
+        return field.supportedByAggsOnlyQuery() && pattern.supportedByAggsOnlyQuery();
     }
 
     @Override
     public boolean resolved() {
-        return source.resolved() && pattern.resolved();
+        return field.resolved() && pattern.resolved();
     }
 
-    protected Pipe replaceChildren(Pipe newSource, Pipe newPattern) {
+    protected Pipe replaceChildren(Pipe newField, Pipe newPattern) {
-        return new StartsWithFunctionPipe(source(), expression(), newSource, newPattern);
+        return new StartsWithFunctionPipe(source(), expression(), newField, newPattern, isCaseSensitive);
     }
 
     @Override
     public final void collectFields(QlSourceBuilder sourceBuilder) {
-        source.collectFields(sourceBuilder);
+        field.collectFields(sourceBuilder);
         pattern.collectFields(sourceBuilder);
     }
 
     @Override
     protected NodeInfo<StartsWithFunctionPipe> info() {
-        return NodeInfo.create(this, StartsWithFunctionPipe::new, expression(), source, pattern);
+        return NodeInfo.create(this, StartsWithFunctionPipe::new, expression(), field, pattern, isCaseSensitive);
     }
 
     @Override
     public StartsWithFunctionProcessor asProcessor() {
-        return new StartsWithFunctionProcessor(source.asProcessor(), pattern.asProcessor());
+        return new StartsWithFunctionProcessor(field.asProcessor(), pattern.asProcessor(), isCaseSensitive);
     }
     
-    public Pipe src() {
+    public Pipe field() {
-        return source;
+        return field;
     }
 
     public Pipe pattern() {
         return pattern;
     }
 
+    public boolean isCaseSensitive() {
+        return isCaseSensitive;
+    }
+
     @Override
     public int hashCode() {
-        return Objects.hash(source, pattern);
+        return Objects.hash(field, pattern, isCaseSensitive);
     }
 
     @Override
@@ -98,7 +104,8 @@ public class StartsWithFunctionPipe extends Pipe {
         }
 
         StartsWithFunctionPipe other = (StartsWithFunctionPipe) obj;
-        return Objects.equals(source, other.source)
+        return Objects.equals(field, other.field)
-                && Objects.equals(pattern, other.pattern);
+                && Objects.equals(pattern, other.pattern)
+                && Objects.equals(isCaseSensitive, other.isCaseSensitive);
     }
 }

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/expression/function/scalar/string/StartsWithFunctionProcessor.java

@@ -3,11 +3,11 @@
  * or more contributor license agreements. Licensed under the Elastic License;
  * you may not use this file except in compliance with the Elastic License.
  */
-package org.elasticsearch.xpack.eql.expression.function.scalar.string;
+package org.elasticsearch.xpack.ql.expression.function.scalar.string;
 
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.xpack.eql.EqlIllegalArgumentException;
+import org.elasticsearch.xpack.ql.QlIllegalArgumentException;
 import org.elasticsearch.xpack.ql.expression.gen.processor.Processor;
 
 import java.io.IOException;
@@ -20,43 +20,51 @@ public class StartsWithFunctionProcessor implements Processor {
 
     private final Processor source;
     private final Processor pattern;
+    private final boolean isCaseSensitive;
 
-    public StartsWithFunctionProcessor(Processor source, Processor pattern) {
+    public StartsWithFunctionProcessor(Processor source, Processor pattern, boolean isCaseSensitive) {
         this.source = source;
         this.pattern = pattern;
+        this.isCaseSensitive = isCaseSensitive;
     }
 
     public StartsWithFunctionProcessor(StreamInput in) throws IOException {
         source = in.readNamedWriteable(Processor.class);
         pattern = in.readNamedWriteable(Processor.class);
+        isCaseSensitive = in.readBoolean();
     }
 
     @Override
     public final void writeTo(StreamOutput out) throws IOException {
         out.writeNamedWriteable(source);
         out.writeNamedWriteable(pattern);
+        out.writeBoolean(isCaseSensitive);
     }
 
     @Override
     public Object process(Object input) {
-        return doProcess(source.process(input), pattern.process(input));
+        return doProcess(source.process(input), pattern.process(input), isCaseSensitive());
     }
 
-    public static Object doProcess(Object source, Object pattern) {
+    public static Object doProcess(Object source, Object pattern, boolean isCaseSensitive) {
         if (source == null) {
             return null;
         }
         if (source instanceof String == false && source instanceof Character == false) {
-            throw new EqlIllegalArgumentException("A string/char is required; received [{}]", source);
+            throw new QlIllegalArgumentException("A string/char is required; received [{}]", source);
         }
         if (pattern == null) {
             return null;
         }
         if (pattern instanceof String == false && pattern instanceof Character == false) {
-            throw new EqlIllegalArgumentException("A string/char is required; received [{}]", pattern);
+            throw new QlIllegalArgumentException("A string/char is required; received [{}]", pattern);
         }
 
-        return source.toString().toLowerCase(Locale.ROOT).startsWith(pattern.toString().toLowerCase(Locale.ROOT));
+        if (isCaseSensitive) {
+            return source.toString().startsWith(pattern.toString());
+        } else {
+            return source.toString().toLowerCase(Locale.ROOT).startsWith(pattern.toString().toLowerCase(Locale.ROOT));
+        }
     }
     
     protected Processor source() {
@@ -67,6 +75,10 @@ public class StartsWithFunctionProcessor implements Processor {
         return pattern;
     }
 
+    protected boolean isCaseSensitive() {
+        return isCaseSensitive;
+    }
+
     @Override
     public boolean equals(Object obj) {
         if (this == obj) {
@@ -79,12 +91,13 @@ public class StartsWithFunctionProcessor implements Processor {
         
         StartsWithFunctionProcessor other = (StartsWithFunctionProcessor) obj;
         return Objects.equals(source(), other.source())
-                && Objects.equals(pattern(), other.pattern());
+                && Objects.equals(pattern(), other.pattern())
+                && Objects.equals(isCaseSensitive(), other.isCaseSensitive());
     }
     
     @Override
     public int hashCode() {
-        return Objects.hash(source(), pattern());
+        return Objects.hash(source(), pattern(), isCaseSensitive());
     }
     
 

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

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.ql.expression.function.scalar.whitelist;
 
 import org.elasticsearch.index.fielddata.ScriptDocValues;
 import org.elasticsearch.xpack.ql.expression.predicate.logical.BinaryLogicProcessor.BinaryLogicOperation;
+import org.elasticsearch.xpack.ql.expression.function.scalar.string.StartsWithFunctionProcessor;
 import org.elasticsearch.xpack.ql.expression.predicate.logical.NotProcessor;
 import org.elasticsearch.xpack.ql.expression.predicate.nulls.CheckNullProcessor.CheckNullOperation;
 import org.elasticsearch.xpack.ql.expression.predicate.operator.arithmetic.DefaultBinaryArithmeticOperation;
@@ -143,4 +144,11 @@ public class InternalQlScriptUtils {
     public static Number sub(Number left, Number right) {
         return (Number) DefaultBinaryArithmeticOperation.SUB.apply(left, right);
     }
+
+    //
+    // String
+    //
+    public static Boolean startsWith(String s, String pattern, Boolean isCaseSensitive) {
+        return (Boolean) StartsWithFunctionProcessor.doProcess(s, pattern, isCaseSensitive);
+    }
 }

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/planner/ExpressionTranslators.java

@@ -12,6 +12,7 @@ import org.elasticsearch.xpack.ql.expression.Expression;
 import org.elasticsearch.xpack.ql.expression.Expressions;
 import org.elasticsearch.xpack.ql.expression.FieldAttribute;
 import org.elasticsearch.xpack.ql.expression.function.scalar.ScalarFunction;
+import org.elasticsearch.xpack.ql.expression.function.scalar.string.StartsWith;
 import org.elasticsearch.xpack.ql.expression.predicate.Range;
 import org.elasticsearch.xpack.ql.expression.predicate.fulltext.MatchQueryPredicate;
 import org.elasticsearch.xpack.ql.expression.predicate.fulltext.MultiMatchQueryPredicate;
@@ -36,6 +37,7 @@ import org.elasticsearch.xpack.ql.querydsl.query.BoolQuery;
 import org.elasticsearch.xpack.ql.querydsl.query.MatchQuery;
 import org.elasticsearch.xpack.ql.querydsl.query.MultiMatchQuery;
 import org.elasticsearch.xpack.ql.querydsl.query.NotQuery;
+import org.elasticsearch.xpack.ql.querydsl.query.PrefixQuery;
 import org.elasticsearch.xpack.ql.querydsl.query.Query;
 import org.elasticsearch.xpack.ql.querydsl.query.QueryStringQuery;
 import org.elasticsearch.xpack.ql.querydsl.query.RangeQuery;
@@ -375,6 +377,16 @@ public final class ExpressionTranslators {
         }
 
         public static Query doTranslate(ScalarFunction f, TranslatorHandler handler) {
+            if (f instanceof StartsWith) {
+                StartsWith sw = (StartsWith) f;
+                if (sw.isCaseSensitive() && sw.field() instanceof FieldAttribute && sw.pattern().foldable()) {
+                    String targetFieldName = handler.nameOf(((FieldAttribute) sw.field()).exactAttribute());
+                    String pattern = (String) sw.pattern().fold();
+
+                    return new PrefixQuery(f.source(), targetFieldName, pattern);
+                }
+            }
+
             return handler.wrapFunctionQuery(f, f, new ScriptQuery(f.source(), f.asScript()));
         }
     }

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/querydsl/query/PrefixQuery.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.ql.querydsl.query;
+
+import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.xpack.ql.tree.Source;
+
+import java.util.Objects;
+
+import static org.elasticsearch.index.query.QueryBuilders.prefixQuery;
+
+public class PrefixQuery extends LeafQuery {
+
+    private final String field, query;
+
+    public PrefixQuery(Source source, String field, String query) {
+        super(source);
+        this.field = field;
+        this.query = query;
+    }
+
+    public String field() {
+        return field;
+    }
+
+    public String query() {
+        return query;
+    }
+
+    @Override
+    public QueryBuilder asBuilder() {
+        return prefixQuery(field, query);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(field, query);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+
+        if (obj == null || getClass() != obj.getClass()) {
+            return false;
+        }
+
+        PrefixQuery other = (PrefixQuery) obj;
+        return Objects.equals(field, other.field)
+                && Objects.equals(query, other.query);
+    }
+
+    @Override
+    protected String innerToString() {
+        return field + ":" + query;
+    }
+}

x-pack/plugin/ql/src/test/java/org/elasticsearch/xpack/ql/expression/function/scalar/string/StartsWithFunctionPipeTests.java

@@ -0,0 +1,126 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.ql.expression.function.scalar.string;
+
+import org.elasticsearch.xpack.ql.TestUtils;
+import org.elasticsearch.xpack.ql.expression.Expression;
+import org.elasticsearch.xpack.ql.expression.function.scalar.FunctionTestUtils.Combinations;
+import org.elasticsearch.xpack.ql.expression.gen.pipeline.Pipe;
+import org.elasticsearch.xpack.ql.tree.AbstractNodeTestCase;
+import org.elasticsearch.xpack.ql.tree.Source;
+
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Function;
+
+import static org.elasticsearch.xpack.ql.expression.Expressions.pipe;
+import static org.elasticsearch.xpack.ql.expression.function.scalar.FunctionTestUtils.randomStringLiteral;
+import static org.elasticsearch.xpack.ql.tree.SourceTests.randomSource;
+
+public class StartsWithFunctionPipeTests extends AbstractNodeTestCase<StartsWithFunctionPipe, Pipe> {
+
+    @Override
+    protected StartsWithFunctionPipe randomInstance() {
+        return randomStartsWithFunctionPipe();
+    }
+    
+    private Expression randomStartsWithFunctionExpression() {
+        return randomStartsWithFunctionPipe().expression();
+    }
+    
+    public static StartsWithFunctionPipe randomStartsWithFunctionPipe() {
+        return (StartsWithFunctionPipe) (new StartsWith(randomSource(),
+                            randomStringLiteral(),
+                            randomStringLiteral(),
+                            TestUtils.randomConfiguration())
+                .makePipe());
+    }
+
+    @Override
+    public void testTransform() {
+        // test transforming only the properties (source, expression),
+        // skipping the children (the two parameters of the binary function) which are tested separately
+        StartsWithFunctionPipe b1 = randomInstance();
+        Expression newExpression = randomValueOtherThan(b1.expression(), () -> randomStartsWithFunctionExpression());
+        StartsWithFunctionPipe newB = new StartsWithFunctionPipe(
+            b1.source(),
+            newExpression,
+            b1.field(),
+            b1.pattern(),
+            b1.isCaseSensitive());
+
+        assertEquals(newB, b1.transformPropertiesOnly(v -> Objects.equals(v, b1.expression()) ? newExpression : v, Expression.class));
+        
+        StartsWithFunctionPipe b2 = randomInstance();
+        Source newLoc = randomValueOtherThan(b2.source(), () -> randomSource());
+        newB = new StartsWithFunctionPipe(
+                newLoc,
+                b2.expression(),
+                b2.field(),
+                b2.pattern(),
+                b2.isCaseSensitive());
+
+        assertEquals(newB,
+                b2.transformPropertiesOnly(v -> Objects.equals(v, b2.source()) ? newLoc : v, Source.class));
+    }
+
+    @Override
+    public void testReplaceChildren() {
+        StartsWithFunctionPipe b = randomInstance();
+        Pipe newField = pipe(((Expression) randomValueOtherThan(b.field(), () -> randomStringLiteral())));
+        Pipe newPattern = pipe(((Expression) randomValueOtherThan(b.pattern(), () -> randomStringLiteral())));
+        
+        StartsWithFunctionPipe newB = new StartsWithFunctionPipe(b.source(), b.expression(), b.field(), b.pattern(), b.isCaseSensitive());
+        StartsWithFunctionPipe transformed = (StartsWithFunctionPipe) newB.replaceChildren(newField, b.pattern());
+        assertEquals(transformed.field(), newField);
+        assertEquals(transformed.source(), b.source());
+        assertEquals(transformed.expression(), b.expression());
+        assertEquals(transformed.pattern(), b.pattern());
+        
+        transformed = (StartsWithFunctionPipe) newB.replaceChildren(b.field(), newPattern);
+        assertEquals(transformed.field(), b.field());
+        assertEquals(transformed.source(), b.source());
+        assertEquals(transformed.expression(), b.expression());
+        assertEquals(transformed.pattern(), newPattern);
+        
+        transformed = (StartsWithFunctionPipe) newB.replaceChildren(newField, newPattern);
+        assertEquals(transformed.field(), newField);
+        assertEquals(transformed.source(), b.source());
+        assertEquals(transformed.expression(), b.expression());
+        assertEquals(transformed.pattern(), newPattern);
+    }
+
+    @Override
+    protected StartsWithFunctionPipe mutate(StartsWithFunctionPipe instance) {
+        List<Function<StartsWithFunctionPipe, StartsWithFunctionPipe>> randoms = new ArrayList<>();
+        for (int i = 1; i < 4; i++) {
+            for (BitSet comb : new Combinations(3, i)) {
+                randoms.add(f -> new StartsWithFunctionPipe(f.source(),
+                        f.expression(),
+                        comb.get(0) ? pipe(((Expression) randomValueOtherThan(f.field(),
+                                () -> randomStringLiteral()))) : f.field(),
+                        comb.get(1) ? pipe(((Expression) randomValueOtherThan(f.pattern(),
+                                () -> randomStringLiteral()))) : f.pattern(),
+                        comb.get(2) ? randomValueOtherThan(f.isCaseSensitive(),
+                                () -> randomBoolean()) : f.isCaseSensitive()));
+            }
+        }
+        
+        return randomFrom(randoms).apply(instance);
+    }
+
+    @Override
+    protected StartsWithFunctionPipe copy(StartsWithFunctionPipe instance) {
+        return new StartsWithFunctionPipe(instance.source(),
+                        instance.expression(),
+                        instance.field(),
+                        instance.pattern(),
+                        instance.isCaseSensitive());
+    }
+}

x-pack/plugin/ql/src/test/java/org/elasticsearch/xpack/ql/expression/function/scalar/string/StartsWithProcessorTests.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.ql.expression.function.scalar.string;
+
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.ql.QlIllegalArgumentException;
+import org.elasticsearch.xpack.ql.expression.Expression;
+import org.elasticsearch.xpack.ql.expression.Literal;
+import org.elasticsearch.xpack.ql.expression.LiteralTests;
+import org.elasticsearch.xpack.ql.session.Configuration;
+import org.elasticsearch.xpack.ql.tree.Source;
+import org.junit.Before;
+
+import java.util.function.Supplier;
+
+import static org.elasticsearch.xpack.ql.TestUtils.randomConfiguration;
+import static org.elasticsearch.xpack.ql.expression.function.scalar.FunctionTestUtils.l;
+import static org.elasticsearch.xpack.ql.tree.Source.EMPTY;
+import static org.elasticsearch.xpack.ql.type.DataTypes.KEYWORD;
+import static org.hamcrest.Matchers.startsWith;
+
+public class StartsWithProcessorTests extends ESTestCase {
+
+    protected boolean isCaseSensitive;
+    protected Configuration config;
+
+    @Before
+    public void setup() {
+        isCaseSensitive = isCaseSensitiveGenerator().get();
+        config = configurationGenerator().get();
+    }
+
+    protected Supplier<Boolean> isCaseSensitiveGenerator() {
+        return () -> true;
+    }
+
+    protected Supplier<Configuration> configurationGenerator() {
+        return () -> randomConfiguration();
+    }
+
+    protected Supplier<StartsWith> startsWithInstantiator(Source source, Expression field, Expression pattern) {
+        return () -> new StartsWith(source, field, pattern, config);
+    }
+
+    public void testStartsWithFunctionWithValidInput() {
+        assertEquals(true, startsWithInstantiator(EMPTY, l("foobarbar"), l("f")).get().makePipe().asProcessor().process(null));
+        assertEquals(false, startsWithInstantiator(EMPTY, l("foobar"), l("bar")).get().makePipe().asProcessor().process(null));
+        assertEquals(false, startsWithInstantiator(EMPTY, l("foo"), l("foobar")).get().makePipe().asProcessor().process(null));
+        assertEquals(true, startsWithInstantiator(EMPTY, l("foobar"), l("")).get().makePipe().asProcessor().process(null));
+        assertEquals(true, startsWithInstantiator(EMPTY, l("foo"), l("foo")).get().makePipe().asProcessor().process(null));
+        assertEquals(!isCaseSensitive, startsWithInstantiator(EMPTY, l("foo"), l("FO")).get().makePipe().asProcessor().process(null));
+        assertEquals(!isCaseSensitive, startsWithInstantiator(EMPTY, l("foo"), l("FOo")).get().makePipe().asProcessor().process(null));
+        assertEquals(true, startsWithInstantiator(EMPTY, l("FOoBar"), l("FOo")).get().makePipe().asProcessor().process(null));
+        assertEquals(true, startsWithInstantiator(EMPTY, l('f'), l('f')).get().makePipe().asProcessor().process(null));
+        assertEquals(false, startsWithInstantiator(EMPTY, l(""), l("bar")).get().makePipe().asProcessor().process(null));
+        assertEquals(null, startsWithInstantiator(EMPTY, l(null), l("bar")).get().makePipe().asProcessor().process(null));
+        assertEquals(null, startsWithInstantiator(EMPTY, l("foo"), l(null)).get().makePipe().asProcessor().process(null));
+        assertEquals(null, startsWithInstantiator(EMPTY, l(null), l(null)).get().makePipe().asProcessor().process(null));
+    }
+    
+    public void testStartsWithFunctionInputsValidation() {
+        QlIllegalArgumentException siae = expectThrows(QlIllegalArgumentException.class,
+                () -> startsWithInstantiator(EMPTY, l(5), l("foo")).get().makePipe().asProcessor().process(null));
+        assertEquals("A string/char is required; received [5]", siae.getMessage());
+        siae = expectThrows(QlIllegalArgumentException.class,
+                () -> startsWithInstantiator(EMPTY, l("bar"), l(false)).get().makePipe().asProcessor().process(null));
+        assertEquals("A string/char is required; received [false]", siae.getMessage());
+    }
+
+    public void testStartsWithFunctionWithRandomInvalidDataType() {
+        Literal literal = randomValueOtherThanMany(v -> v.dataType() == KEYWORD, () -> LiteralTests.randomLiteral());
+        QlIllegalArgumentException siae = expectThrows(QlIllegalArgumentException.class,
+                () -> startsWithInstantiator(EMPTY, literal, l("foo")).get().makePipe().asProcessor().process(null));
+        assertThat(siae.getMessage(), startsWith("A string/char is required; received"));
+        siae = expectThrows(QlIllegalArgumentException.class,
+                () -> startsWithInstantiator(EMPTY, l("foo"), literal).get().makePipe().asProcessor().process(null));
+        assertThat(siae.getMessage(), startsWith("A string/char is required; received"));
+    }
+}

x-pack/plugin/sql/qa/src/main/resources/command.csv-spec

@@ -145,6 +145,7 @@ REPLACE          |SCALAR
 RIGHT            |SCALAR         
 RTRIM            |SCALAR         
 SPACE            |SCALAR         
+STARTS_WITH      |SCALAR
 SUBSTRING        |SCALAR         
 UCASE            |SCALAR
 CAST             |SCALAR

x-pack/plugin/sql/qa/src/main/resources/docs/docs.csv-spec

@@ -341,6 +341,7 @@ REPLACE          |SCALAR
 RIGHT            |SCALAR         
 RTRIM            |SCALAR         
 SPACE            |SCALAR         
+STARTS_WITH      |SCALAR
 SUBSTRING        |SCALAR         
 UCASE            |SCALAR
 CAST             |SCALAR
@@ -1797,6 +1798,26 @@ SELECT SPACE(3);
 // end::stringSpace
 ;
 
+stringStartsWithTrue
+// tag::stringStartsWithTrue
+SELECT STARTS_WITH('Elasticsearch', 'Elastic');
+
+STARTS_WITH('Elasticsearch', 'Elastic')
+--------------------------------
+true    
+// end::stringStartsWithTrue
+;
+
+stringStartsWithFalse
+// tag::stringStartsWithFalse
+SELECT STARTS_WITH('Elasticsearch', 'ELASTIC');
+
+STARTS_WITH('Elasticsearch', 'ELASTIC')
+--------------------------------
+false   
+// end::stringStartsWithFalse
+;
+
 stringSubString
 // tag::stringSubString
 SELECT SUBSTRING('Elasticsearch', 0, 7);

x-pack/plugin/sql/qa/src/main/resources/functions.csv-spec

@@ -476,6 +476,115 @@ AlejandRo      |1
 ;
 
 
+startsWithInline1
+SELECT STARTS_WITH('Elasticsearch', 'Elastic') stwith;
+
+  stwith:b   
+---------------
+true
+;
+
+startsWithInline2
+SELECT STARTS_WITH('Elasticsearch', 'elastic') stwith;
+
+  stwith:b   
+---------------
+false
+;
+
+selectStartsWith_WithThreeConditions
+SELECT "first_name" FROM "test_emp" WHERE STARTS_WITH("first_name", 'A') OR STARTS_WITH("first_name", 'Br') OR STARTS_WITH(LCASE("first_name"), LCASE('X')) ORDER BY "first_name" DESC;
+
+  first_name:s   
+---------------
+Xinglin        
+Brendon        
+Breannda       
+Arumugam       
+Anoosh         
+Anneke         
+Amabile        
+Alejandro      
+;
+
+selectStartsWith_WithGroupByAndOrderBy
+SELECT STARTS_WITH("first_name", 'A') st, COUNT(*) count FROM "test_emp" WHERE st IS NOT NULL GROUP BY 1 ORDER BY 1;
+
+      st:b       |     count:l     
+-----------------+---------------
+false            |85             
+true             |5              
+;
+
+selectStartsWithIsNull
+SELECT STARTS_WITH("first_name", 'A') IS NULL stwith, first_name FROM test_emp ORDER BY stwith DESC LIMIT 15;
+
+   stwith:b    |  first_name:s   
+---------------+---------------
+true           |null           
+true           |null           
+true           |null           
+true           |null           
+true           |null           
+true           |null           
+true           |null           
+true           |null           
+true           |null           
+true           |null           
+false          |Georgi         
+false          |Bezalel        
+false          |Parto          
+false          |Chirstian      
+false          |Kyoichi        
+;
+
+selectStartsWith_WithWhereAndGroupBy
+SELECT STARTS_WITH("first_name", 'A') st, COUNT(*) count FROM "test_emp" WHERE st IS NOT NULL GROUP BY STARTS_WITH("first_name", 'A');
+
+      st:b       |     count:l     
+-----------------+---------------
+false            |85             
+true             |5              
+;
+
+selectStartsWith_WithWhereAndGroupByAndOrderBy
+SELECT MAX("salary") s, STARTS_WITH("first_name", 'A') st, MIN("salary") a FROM "test_emp" WHERE st IS NOT NULL GROUP BY STARTS_WITH("first_name", 'A') ORDER BY st;
+
+       s:i     |      st:b       |       a:i       
+---------------+-----------------+---------------
+74999          |false            |25324        
+66817          |true             |38645        
+;
+
+selectComplexStartsWith_WithWhereAndNestedFunction
+SELECT MAX("salary") max, CONCAT(CASE WHEN STARTS_WITH(LCASE("first_name"), 'a') THEN 'elasticsearch' ELSE 'search' END, '@elastic') concat_st, STARTS_WITH(LCASE("first_name"), 'a') st, MIN("salary") min, COUNT(*) count FROM test_emp WHERE STARTS_WITH(LCASE("first_name"), 'a') IS NOT NULL GROUP BY STARTS_WITH(LCASE("first_name"), 'a');
+
+     max:i     |      concat_st:s      |      st:b       |      min:i    |     count:l     
+---------------+-----------------------+-----------------+---------------+---------------
+74999          |search@elastic         |false            |25324          |85             
+66817          |elasticsearch@elastic  |true             |38645          |5              
+;
+
+selectStartsWith_WithTwoColumns
+SELECT "first_name", "last_name", "gender" FROM "test_emp" WHERE STARTS_WITH("first_name", "gender") OR STARTS_WITH("last_name", "gender") ORDER BY "gender";
+
+ first_name:s  |  last_name:s  |   gender:s     
+---------------+---------------+---------------
+Sudharsan      |Flasterstein   |F              
+Kyoichi        |Maliniak       |M              
+Mayuko         |Warwick        |M              
+null           |Merlo          |M              
+null           |Makrucki       |M              
+Moss           |Shanbhogue     |M              
+Mayumi         |Schueller      |M              
+Berhard        |McFarlin       |M              
+Shir           |McClurg        |M              
+Mona           |Azuma          |M              
+Kenroku        |Malabarba      |M              
+Hilari         |Morton         |M              
+Jayson         |Mandell        |M              
+;
+
 checkColumnNameWithNestedArithmeticFunctionCallsOnTableColumn
 SELECT CHAR(emp_no % 10000) AS c FROM "test_emp" WHERE emp_no > 10064 ORDER BY emp_no LIMIT 1;
 

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/execution/PlanExecutor.java

@@ -22,7 +22,7 @@ import org.elasticsearch.xpack.sql.plan.physical.LocalExec;
 import org.elasticsearch.xpack.sql.planner.Planner;
 import org.elasticsearch.xpack.sql.planner.PlanningException;
 import org.elasticsearch.xpack.sql.proto.SqlTypedParamValue;
-import org.elasticsearch.xpack.sql.session.Configuration;
+import org.elasticsearch.xpack.sql.session.SqlConfiguration;
 import org.elasticsearch.xpack.sql.session.Cursor;
 import org.elasticsearch.xpack.sql.session.Cursor.Page;
 import org.elasticsearch.xpack.sql.session.SqlSession;
@@ -62,11 +62,12 @@ public class PlanExecutor {
         this.planner = new Planner();
     }
 
-    private SqlSession newSession(Configuration cfg) {
+    private SqlSession newSession(SqlConfiguration cfg) {
         return new SqlSession(cfg, client, functionRegistry, indexResolver, preAnalyzer, verifier, optimizer, planner, this);
     }
 
-    public void searchSource(Configuration cfg, String sql, List<SqlTypedParamValue> params, ActionListener<SearchSourceBuilder> listener) {
+    public void searchSource(SqlConfiguration cfg, String sql, List<SqlTypedParamValue> params,
+            ActionListener<SearchSourceBuilder> listener) {
         metrics.translate();
 
         newSession(cfg).sqlExecutable(sql, params, wrap(exec -> {
@@ -91,7 +92,7 @@ public class PlanExecutor {
         }, listener::onFailure));
     }
 
-    public void sql(Configuration cfg, String sql, List<SqlTypedParamValue> params, ActionListener<Page> listener) {
+    public void sql(SqlConfiguration cfg, String sql, List<SqlTypedParamValue> params, ActionListener<Page> listener) {
         QueryMetric metric = QueryMetric.from(cfg.mode(), cfg.clientId());
         metrics.total(metric);
 
@@ -101,7 +102,7 @@ public class PlanExecutor {
         }));
     }
 
-    public void nextPage(Configuration cfg, Cursor cursor, ActionListener<Page> listener) {
+    public void nextPage(SqlConfiguration cfg, Cursor cursor, ActionListener<Page> listener) {
         QueryMetric metric = QueryMetric.from(cfg.mode(), cfg.clientId());
         metrics.total(metric);
         metrics.paging(metric);
@@ -112,7 +113,7 @@ public class PlanExecutor {
         }));
     }
 
-    public void cleanCursor(Configuration cfg, Cursor cursor, ActionListener<Boolean> listener) {
+    public void cleanCursor(SqlConfiguration cfg, Cursor cursor, ActionListener<Boolean> listener) {
         cursor.clear(cfg, client, listener);
     }
     

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/execution/search/CompositeAggCursor.java

@@ -27,7 +27,7 @@ import org.elasticsearch.xpack.ql.type.Schema;
 import org.elasticsearch.xpack.ql.util.StringUtils;
 import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
 import org.elasticsearch.xpack.sql.querydsl.agg.Aggs;
-import org.elasticsearch.xpack.sql.session.Configuration;
+import org.elasticsearch.xpack.sql.session.SqlConfiguration;
 import org.elasticsearch.xpack.sql.session.Cursor;
 import org.elasticsearch.xpack.sql.session.Rows;
 
@@ -118,7 +118,7 @@ public class CompositeAggCursor implements Cursor {
     }
 
     @Override
-    public void nextPage(Configuration cfg, Client client, NamedWriteableRegistry registry, ActionListener<Page> listener) {
+    public void nextPage(SqlConfiguration cfg, Client client, NamedWriteableRegistry registry, ActionListener<Page> listener) {
         SearchSourceBuilder q;
         try {
             q = deserializeQuery(registry, nextQuery);
@@ -268,7 +268,7 @@ public class CompositeAggCursor implements Cursor {
 
 
     @Override
-    public void clear(Configuration cfg, Client client, ActionListener<Boolean> listener) {
+    public void clear(SqlConfiguration cfg, Client client, ActionListener<Boolean> listener) {
         listener.onResponse(true);
     }
 

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/execution/search/Querier.java

@@ -58,7 +58,7 @@ import org.elasticsearch.xpack.sql.querydsl.container.QueryContainer;
 import org.elasticsearch.xpack.sql.querydsl.container.ScriptFieldRef;
 import org.elasticsearch.xpack.sql.querydsl.container.SearchHitFieldRef;
 import org.elasticsearch.xpack.sql.querydsl.container.TopHitsAggRef;
-import org.elasticsearch.xpack.sql.session.Configuration;
+import org.elasticsearch.xpack.sql.session.SqlConfiguration;
 import org.elasticsearch.xpack.sql.session.Cursor;
 import org.elasticsearch.xpack.sql.session.Cursor.Page;
 import org.elasticsearch.xpack.sql.session.ListCursor;
@@ -90,7 +90,7 @@ public class Querier {
     private static final Logger log = LogManager.getLogger(Querier.class);
 
     private final PlanExecutor planExecutor;
-    private final Configuration cfg;
+    private final SqlConfiguration cfg;
     private final TimeValue keepAlive, timeout;
     private final int size;
     private final Client client;
@@ -297,7 +297,7 @@ public class Querier {
             }
         });
 
-        ImplicitGroupActionListener(ActionListener<Page> listener, Client client, Configuration cfg, List<Attribute> output,
+        ImplicitGroupActionListener(ActionListener<Page> listener, Client client, SqlConfiguration cfg, List<Attribute> output,
                 QueryContainer query, SearchRequest request) {
             super(listener, client, cfg, output, query, request);
         }
@@ -354,7 +354,7 @@ public class Querier {
 
         private final boolean isPivot;
 
-        CompositeActionListener(ActionListener<Page> listener, Client client, Configuration cfg, List<Attribute> output,
+        CompositeActionListener(ActionListener<Page> listener, Client client, SqlConfiguration cfg, List<Attribute> output,
                 QueryContainer query, SearchRequest request) {
             super(listener, client, cfg, output, query, request);
 
@@ -385,8 +385,8 @@ public class Querier {
         final SearchRequest request;
         final BitSet mask;
 
-        BaseAggActionListener(ActionListener<Page> listener, Client client, Configuration cfg, List<Attribute> output, QueryContainer query,
+        BaseAggActionListener(ActionListener<Page> listener, Client client, SqlConfiguration cfg, List<Attribute> output,
-                SearchRequest request) {
+                QueryContainer query, SearchRequest request) {
             super(listener, client, cfg, output);
 
             this.query = query;
@@ -455,7 +455,7 @@ public class Querier {
         private final BitSet mask;
         private final boolean multiValueFieldLeniency;
 
-        ScrollActionListener(ActionListener<Page> listener, Client client, Configuration cfg, List<Attribute> output,
+        ScrollActionListener(ActionListener<Page> listener, Client client, SqlConfiguration cfg, List<Attribute> output,
                 QueryContainer query) {
             super(listener, client, cfg, output);
             this.query = query;
@@ -524,11 +524,11 @@ public class Querier {
         final ActionListener<Page> listener;
 
         final Client client;
-        final Configuration cfg;
+        final SqlConfiguration cfg;
         final TimeValue keepAlive;
         final Schema schema;
 
-        BaseActionListener(ActionListener<Page> listener, Client client, Configuration cfg, List<Attribute> output) {
+        BaseActionListener(ActionListener<Page> listener, Client client, SqlConfiguration cfg, List<Attribute> output) {
             this.listener = listener;
 
             this.client = client;

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/execution/search/ScrollCursor.java

@@ -21,7 +21,7 @@ import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.xpack.ql.execution.search.extractor.HitExtractor;
 import org.elasticsearch.xpack.ql.type.Schema;
-import org.elasticsearch.xpack.sql.session.Configuration;
+import org.elasticsearch.xpack.sql.session.SqlConfiguration;
 import org.elasticsearch.xpack.sql.session.Cursor;
 import org.elasticsearch.xpack.sql.session.Rows;
 
@@ -90,7 +90,7 @@ public class ScrollCursor implements Cursor {
         return limit;
     }
     @Override
-    public void nextPage(Configuration cfg, Client client, NamedWriteableRegistry registry, ActionListener<Page> listener) {
+    public void nextPage(SqlConfiguration cfg, Client client, NamedWriteableRegistry registry, ActionListener<Page> listener) {
         if (log.isTraceEnabled()) {
             log.trace("About to execute scroll query {}", scrollId);
         }
@@ -105,7 +105,7 @@ public class ScrollCursor implements Cursor {
     }
 
     @Override
-    public void clear(Configuration cfg, Client client, ActionListener<Boolean> listener) {
+    public void clear(SqlConfiguration cfg, Client client, ActionListener<Boolean> listener) {
         cleanCursor(client, scrollId, wrap(
                         clearScrollResponse -> listener.onResponse(clearScrollResponse.isSucceeded()),
                         listener::onFailure));

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/SqlFunctionRegistry.java

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.sql.expression.function;
 import org.elasticsearch.xpack.ql.expression.function.FunctionDefinition;
 import org.elasticsearch.xpack.ql.expression.function.FunctionRegistry;
 import org.elasticsearch.xpack.ql.expression.function.aggregate.Count;
+import org.elasticsearch.xpack.ql.expression.function.scalar.string.StartsWith;
 import org.elasticsearch.xpack.sql.expression.function.aggregate.Avg;
 import org.elasticsearch.xpack.sql.expression.function.aggregate.First;
 import org.elasticsearch.xpack.sql.expression.function.aggregate.Kurtosis;
@@ -238,6 +239,7 @@ public class SqlFunctionRegistry extends FunctionRegistry {
                 def(Right.class, Right::new, "RIGHT"),
                 def(RTrim.class, RTrim::new, "RTRIM"),
                 def(Space.class, Space::new, "SPACE"),
+                def(StartsWith.class, StartsWith::new, "STARTS_WITH"),
                 def(Substring.class, Substring::new, "SUBSTRING"),
                 def(UCase.class, UCase::new, "UCASE")
             },
@@ -266,4 +268,5 @@ public class SqlFunctionRegistry extends FunctionRegistry {
             }
         };
     }
+
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/Database.java

@@ -11,7 +11,7 @@ import org.elasticsearch.xpack.ql.tree.NodeInfo;
 import org.elasticsearch.xpack.ql.tree.Source;
 import org.elasticsearch.xpack.ql.type.DataTypes;
 
-public class Database extends ConfigurationFunction {
+public class Database extends SqlConfigurationFunction {
 
     public Database(Source source, Configuration configuration) {
         super(source, configuration, DataTypes.KEYWORD);

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/Processors.java

@@ -7,6 +7,7 @@ package org.elasticsearch.xpack.sql.expression.function.scalar;
 
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry.Entry;
+import org.elasticsearch.xpack.ql.expression.function.scalar.string.StartsWithFunctionProcessor;
 import org.elasticsearch.xpack.ql.expression.gen.processor.Processor;
 import org.elasticsearch.xpack.ql.expression.predicate.nulls.CheckNullProcessor;
 import org.elasticsearch.xpack.ql.expression.predicate.operator.arithmetic.BinaryArithmeticOperation;
@@ -101,6 +102,7 @@ public final class Processors {
         entries.add(new Entry(Processor.class, LocateFunctionProcessor.NAME, LocateFunctionProcessor::new));
         entries.add(new Entry(Processor.class, ReplaceFunctionProcessor.NAME, ReplaceFunctionProcessor::new));
         entries.add(new Entry(Processor.class, SubstringFunctionProcessor.NAME, SubstringFunctionProcessor::new));
+        entries.add(new Entry(Processor.class, StartsWithFunctionProcessor.NAME, StartsWithFunctionProcessor::new));
         // geo
         entries.add(new Entry(Processor.class, GeoProcessor.NAME, GeoProcessor::new));
         entries.add(new Entry(Processor.class, StWkttosqlProcessor.NAME, StWkttosqlProcessor::new));

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/SqlConfigurationFunction.java

@@ -8,7 +8,7 @@ package org.elasticsearch.xpack.sql.expression.function.scalar;
 
 import org.elasticsearch.xpack.ql.expression.Expression;
 import org.elasticsearch.xpack.ql.expression.Nullability;
-import org.elasticsearch.xpack.ql.expression.function.scalar.ScalarFunction;
+import org.elasticsearch.xpack.ql.expression.function.scalar.ConfigurationFunction;
 import org.elasticsearch.xpack.ql.expression.gen.script.ScriptTemplate;
 import org.elasticsearch.xpack.ql.session.Configuration;
 import org.elasticsearch.xpack.ql.tree.Source;
@@ -17,14 +17,14 @@ import org.elasticsearch.xpack.ql.type.DataType;
 import java.util.List;
 import java.util.Objects;
 
-public abstract class ConfigurationFunction extends ScalarFunction {
+import static java.util.Collections.emptyList;
+
+public abstract class SqlConfigurationFunction extends ConfigurationFunction {
 
-    private final Configuration configuration;
     private final DataType dataType;
 
-    protected ConfigurationFunction(Source source, Configuration configuration, DataType dataType) {
+    protected SqlConfigurationFunction(Source source, Configuration configuration, DataType dataType) {
-        super(source);
+        super(source, emptyList(), configuration);
-        this.configuration = configuration;
         this.dataType = dataType;
     }
 
@@ -33,10 +33,6 @@ public abstract class ConfigurationFunction extends ScalarFunction {
         throw new UnsupportedOperationException("this node doesn't have any children");
     }
 
-    public Configuration configuration() {
-        return configuration;
-    }
-
     @Override
     public DataType dataType() {
         return dataType;
@@ -67,6 +63,6 @@ public abstract class ConfigurationFunction extends ScalarFunction {
 
     @Override
     public boolean equals(Object obj) {
-        return super.equals(obj) && Objects.equals(fold(), ((ConfigurationFunction) obj).fold());
+        return super.equals(obj) && Objects.equals(fold(), ((SqlConfigurationFunction) obj).fold());
     }
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/User.java

@@ -11,7 +11,7 @@ import org.elasticsearch.xpack.ql.tree.NodeInfo;
 import org.elasticsearch.xpack.ql.tree.Source;
 import org.elasticsearch.xpack.ql.type.DataTypes;
 
-public class User extends ConfigurationFunction {
+public class User extends SqlConfigurationFunction {
 
     public User(Source source, Configuration configuration) {
         super(source, configuration, DataTypes.KEYWORD);

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/CurrentFunction.java

@@ -9,12 +9,12 @@ package org.elasticsearch.xpack.sql.expression.function.scalar.datetime;
 import org.elasticsearch.xpack.ql.session.Configuration;
 import org.elasticsearch.xpack.ql.tree.Source;
 import org.elasticsearch.xpack.ql.type.DataType;
-import org.elasticsearch.xpack.sql.expression.function.scalar.ConfigurationFunction;
+import org.elasticsearch.xpack.sql.expression.function.scalar.SqlConfigurationFunction;
 
 import java.time.temporal.Temporal;
 import java.util.Objects;
 
-abstract class CurrentFunction<T extends Temporal> extends ConfigurationFunction {
+abstract class CurrentFunction<T extends Temporal> extends SqlConfigurationFunction {
 
     private final T current;
 

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/plugin/TextFormatterCursor.java

@@ -11,7 +11,7 @@ import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.xpack.sql.action.BasicFormatter;
-import org.elasticsearch.xpack.sql.session.Configuration;
+import org.elasticsearch.xpack.sql.session.SqlConfiguration;
 import org.elasticsearch.xpack.sql.session.Cursor;
 
 import java.io.IOException;
@@ -48,7 +48,7 @@ public class TextFormatterCursor implements Cursor {
     }
 
     @Override
-    public void nextPage(Configuration cfg, Client client, NamedWriteableRegistry registry, ActionListener<Page> listener) {
+    public void nextPage(SqlConfiguration cfg, Client client, NamedWriteableRegistry registry, ActionListener<Page> listener) {
         // keep wrapping the text formatter
         delegate.nextPage(cfg, client, registry,
                 wrap(p -> {
@@ -58,7 +58,7 @@ public class TextFormatterCursor implements Cursor {
     }
 
     @Override
-    public void clear(Configuration cfg, Client client, ActionListener<Boolean> listener) {
+    public void clear(SqlConfiguration cfg, Client client, ActionListener<Boolean> listener) {
         delegate.clear(cfg, client, listener);
     }
 

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/plugin/TransportSqlClearCursorAction.java

@@ -16,7 +16,7 @@ import org.elasticsearch.xpack.sql.action.SqlClearCursorRequest;
 import org.elasticsearch.xpack.sql.action.SqlClearCursorResponse;
 import org.elasticsearch.xpack.sql.execution.PlanExecutor;
 import org.elasticsearch.xpack.sql.proto.Protocol;
-import org.elasticsearch.xpack.sql.session.Configuration;
+import org.elasticsearch.xpack.sql.session.SqlConfiguration;
 import org.elasticsearch.xpack.sql.session.Cursor;
 import org.elasticsearch.xpack.sql.session.Cursors;
 import org.elasticsearch.xpack.sql.util.DateUtils;
@@ -45,7 +45,7 @@ public class TransportSqlClearCursorAction extends HandledTransportAction<SqlCle
             ActionListener<SqlClearCursorResponse> listener) {
         Cursor cursor = Cursors.decodeFromStringWithZone(request.getCursor()).v1();
         planExecutor.cleanCursor(
-                new Configuration(DateUtils.UTC, Protocol.FETCH_SIZE, Protocol.REQUEST_TIMEOUT, Protocol.PAGE_TIMEOUT, null,
+                new SqlConfiguration(DateUtils.UTC, Protocol.FETCH_SIZE, Protocol.REQUEST_TIMEOUT, Protocol.PAGE_TIMEOUT, null,
                         request.mode(), StringUtils.EMPTY, StringUtils.EMPTY, StringUtils.EMPTY, Protocol.FIELD_MULTI_VALUE_LENIENCY,
                         Protocol.INDEX_INCLUDE_FROZEN),
                 cursor, ActionListener.wrap(

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/plugin/TransportSqlQueryAction.java

@@ -26,7 +26,7 @@ import org.elasticsearch.xpack.sql.action.SqlQueryResponse;
 import org.elasticsearch.xpack.sql.execution.PlanExecutor;
 import org.elasticsearch.xpack.sql.proto.ColumnInfo;
 import org.elasticsearch.xpack.sql.proto.Mode;
-import org.elasticsearch.xpack.sql.session.Configuration;
+import org.elasticsearch.xpack.sql.session.SqlConfiguration;
 import org.elasticsearch.xpack.sql.session.Cursor;
 import org.elasticsearch.xpack.sql.session.Cursor.Page;
 import org.elasticsearch.xpack.sql.session.Cursors;
@@ -75,7 +75,7 @@ public class TransportSqlQueryAction extends HandledTransportAction<SqlQueryRequ
                                  String username, String clusterName) {
         // The configuration is always created however when dealing with the next page, only the timeouts are relevant
         // the rest having default values (since the query is already created)
-        Configuration cfg = new Configuration(request.zoneId(), request.fetchSize(), request.requestTimeout(), request.pageTimeout(),
+        SqlConfiguration cfg = new SqlConfiguration(request.zoneId(), request.fetchSize(), request.requestTimeout(), request.pageTimeout(),
                 request.filter(), request.mode(), request.clientId(), username, clusterName, request.fieldMultiValueLeniency(),
                 request.indexIncludeFrozen());
 

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/plugin/TransportSqlTranslateAction.java

@@ -21,7 +21,7 @@ import org.elasticsearch.xpack.sql.action.SqlTranslateRequest;
 import org.elasticsearch.xpack.sql.action.SqlTranslateResponse;
 import org.elasticsearch.xpack.sql.execution.PlanExecutor;
 import org.elasticsearch.xpack.sql.proto.Protocol;
-import org.elasticsearch.xpack.sql.session.Configuration;
+import org.elasticsearch.xpack.sql.session.SqlConfiguration;
 
 import static org.elasticsearch.xpack.sql.plugin.Transports.clusterName;
 import static org.elasticsearch.xpack.sql.plugin.Transports.username;
@@ -52,7 +52,7 @@ public class TransportSqlTranslateAction extends HandledTransportAction<SqlTrans
     protected void doExecute(Task task, SqlTranslateRequest request, ActionListener<SqlTranslateResponse> listener) {
         sqlLicenseChecker.checkIfSqlAllowed(request.mode());
 
-        Configuration cfg = new Configuration(request.zoneId(), request.fetchSize(),
+        SqlConfiguration cfg = new SqlConfiguration(request.zoneId(), request.fetchSize(),
                 request.requestTimeout(), request.pageTimeout(), request.filter(),
                 request.mode(), request.clientId(),
                 username(securityContext), clusterName(clusterService), Protocol.FIELD_MULTI_VALUE_LENIENCY,

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/session/Cursor.java

@@ -42,10 +42,10 @@ public interface Cursor extends NamedWriteable {
     /**
      * Request the next page of data.
      */
-    void nextPage(Configuration cfg, Client client, NamedWriteableRegistry registry, ActionListener<Page> listener);
+    void nextPage(SqlConfiguration cfg, Client client, NamedWriteableRegistry registry, ActionListener<Page> listener);
 
     /**
      *  Cleans the resources associated with the cursor
      */
-    void clear(Configuration cfg, Client client, ActionListener<Boolean> listener);
+    void clear(SqlConfiguration cfg, Client client, ActionListener<Boolean> listener);
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/session/EmptyCursor.java

@@ -32,12 +32,12 @@ class EmptyCursor implements Cursor {
     }
 
     @Override
-    public void nextPage(Configuration cfg, Client client, NamedWriteableRegistry registry, ActionListener<Page> listener) {
+    public void nextPage(SqlConfiguration cfg, Client client, NamedWriteableRegistry registry, ActionListener<Page> listener) {
         throw new SqlIllegalArgumentException("there is no next page");
     }
 
     @Override
-    public void clear(Configuration cfg, Client client, ActionListener<Boolean> listener) {
+    public void clear(SqlConfiguration cfg, Client client, ActionListener<Boolean> listener) {
         // There is nothing to clean
         listener.onResponse(false);
     }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/session/ListCursor.java

@@ -83,12 +83,12 @@ public class ListCursor implements Cursor {
     }
     
     @Override
-    public void nextPage(Configuration cfg, Client client, NamedWriteableRegistry registry, ActionListener<Page> listener) {
+    public void nextPage(SqlConfiguration cfg, Client client, NamedWriteableRegistry registry, ActionListener<Page> listener) {
         listener.onResponse(of(Schema.EMPTY, data, pageSize, columnCount));
     }
 
     @Override
-    public void clear(Configuration cfg, Client client, ActionListener<Boolean> listener) {
+    public void clear(SqlConfiguration cfg, Client client, ActionListener<Boolean> listener) {
         listener.onResponse(true);
     }
 

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/session/SqlConfiguration.java

@@ -13,7 +13,7 @@ import org.elasticsearch.xpack.sql.proto.Mode;
 import java.time.ZoneId;
 
 // Typed object holding properties for a given query
-public class Configuration extends org.elasticsearch.xpack.ql.session.Configuration {
+public class SqlConfiguration extends org.elasticsearch.xpack.ql.session.Configuration {
     
     private final int pageSize;
     private final TimeValue requestTimeout;
@@ -26,7 +26,7 @@ public class Configuration extends org.elasticsearch.xpack.ql.session.Configurat
     @Nullable
     private QueryBuilder filter;
 
-    public Configuration(ZoneId zi, int pageSize, TimeValue requestTimeout, TimeValue pageTimeout, QueryBuilder filter,
+    public SqlConfiguration(ZoneId zi, int pageSize, TimeValue requestTimeout, TimeValue pageTimeout, QueryBuilder filter,
                          Mode mode, String clientId,
                          String username, String clusterName,
                          boolean multiValueFieldLeniency,

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/session/SqlSession.java

@@ -45,9 +45,9 @@ public class SqlSession implements Session {
     private final Planner planner;
     private final PlanExecutor planExecutor;
     
-    private final Configuration configuration;
+    private final SqlConfiguration configuration;
 
-    public SqlSession(Configuration configuration, Client client, FunctionRegistry functionRegistry,
+    public SqlSession(SqlConfiguration configuration, Client client, FunctionRegistry functionRegistry,
             IndexResolver indexResolver,
             PreAnalyzer preAnalyzer,
             Verifier verifier,
@@ -172,7 +172,7 @@ public class SqlSession implements Session {
         }
     }
 
-    public Configuration configuration() {
+    public SqlConfiguration configuration() {
         return configuration;
     }
 }

x-pack/plugin/sql/src/main/resources/org/elasticsearch/xpack/sql/plugin/sql_whitelist.txt

@@ -29,6 +29,11 @@ class org.elasticsearch.xpack.ql.expression.function.scalar.whitelist.InternalQl
   double nullSafeSortNumeric(Number)
   String nullSafeSortString(Object)
 
+#
+# ASCII Functions
+#
+  Boolean startsWith(String, String, Boolean)
+
 #
 # Comparison
 #

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/SqlTestUtils.java

@@ -11,7 +11,7 @@ import org.elasticsearch.xpack.ql.expression.Literal;
 import org.elasticsearch.xpack.ql.tree.Source;
 import org.elasticsearch.xpack.sql.proto.Mode;
 import org.elasticsearch.xpack.sql.proto.Protocol;
-import org.elasticsearch.xpack.sql.session.Configuration;
+import org.elasticsearch.xpack.sql.session.SqlConfiguration;
 import org.elasticsearch.xpack.sql.type.SqlDataTypes;
 import org.elasticsearch.xpack.sql.util.DateUtils;
 
@@ -34,7 +34,7 @@ public final class SqlTestUtils {
 
     private SqlTestUtils() {}
 
-    public static final Configuration TEST_CFG = new Configuration(DateUtils.UTC, Protocol.FETCH_SIZE,
+    public static final SqlConfiguration TEST_CFG = new SqlConfiguration(DateUtils.UTC, Protocol.FETCH_SIZE,
             Protocol.REQUEST_TIMEOUT, Protocol.PAGE_TIMEOUT, null, Mode.PLAIN,
             null, null, null, false, false);
 
@@ -51,8 +51,8 @@ public final class SqlTestUtils {
         return ZonedDateTime.now(Clock.tick(Clock.system(DateUtils.UTC), Duration.ofMillis(1)));
     }
 
-    public static Configuration randomConfiguration() {
+    public static SqlConfiguration randomConfiguration() {
-        return new Configuration(randomZone(),
+        return new SqlConfiguration(randomZone(),
                 randomIntBetween(0,  1000),
                 new TimeValue(randomNonNegativeLong()),
                 new TimeValue(randomNonNegativeLong()),
@@ -65,8 +65,8 @@ public final class SqlTestUtils {
                 randomBoolean());
     }
 
-    public static Configuration randomConfiguration(ZoneId providedZoneId) {
+    public static SqlConfiguration randomConfiguration(ZoneId providedZoneId) {
-        return new Configuration(providedZoneId,
+        return new SqlConfiguration(providedZoneId,
                 randomIntBetween(0,  1000),
                 new TimeValue(randomNonNegativeLong()),
                 new TimeValue(randomNonNegativeLong()),

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/expression/function/scalar/DatabaseFunctionTests.java

@@ -18,7 +18,7 @@ import org.elasticsearch.xpack.sql.expression.function.SqlFunctionRegistry;
 import org.elasticsearch.xpack.sql.parser.SqlParser;
 import org.elasticsearch.xpack.sql.proto.Mode;
 import org.elasticsearch.xpack.sql.proto.Protocol;
-import org.elasticsearch.xpack.sql.session.Configuration;
+import org.elasticsearch.xpack.sql.session.SqlConfiguration;
 import org.elasticsearch.xpack.sql.stats.Metrics;
 import org.elasticsearch.xpack.sql.types.SqlTypesTests;
 import org.elasticsearch.xpack.sql.util.DateUtils;
@@ -30,7 +30,7 @@ public class DatabaseFunctionTests extends ESTestCase {
         SqlParser parser = new SqlParser();
         EsIndex test = new EsIndex("test", SqlTypesTests.loadMapping("mapping-basic.json", true));
         Analyzer analyzer = new Analyzer(
-                new Configuration(DateUtils.UTC, Protocol.FETCH_SIZE, Protocol.REQUEST_TIMEOUT,
+                new SqlConfiguration(DateUtils.UTC, Protocol.FETCH_SIZE, Protocol.REQUEST_TIMEOUT,
                                   Protocol.PAGE_TIMEOUT, null,
                                   randomFrom(Mode.values()), randomAlphaOfLength(10),
                                   null, clusterName, randomBoolean(), randomBoolean()),

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/expression/function/scalar/UserFunctionTests.java

@@ -18,7 +18,7 @@ import org.elasticsearch.xpack.sql.expression.function.SqlFunctionRegistry;
 import org.elasticsearch.xpack.sql.parser.SqlParser;
 import org.elasticsearch.xpack.sql.proto.Mode;
 import org.elasticsearch.xpack.sql.proto.Protocol;
-import org.elasticsearch.xpack.sql.session.Configuration;
+import org.elasticsearch.xpack.sql.session.SqlConfiguration;
 import org.elasticsearch.xpack.sql.stats.Metrics;
 import org.elasticsearch.xpack.sql.types.SqlTypesTests;
 import org.elasticsearch.xpack.sql.util.DateUtils;
@@ -29,7 +29,7 @@ public class UserFunctionTests extends ESTestCase {
         SqlParser parser = new SqlParser();
         EsIndex test = new EsIndex("test", SqlTypesTests.loadMapping("mapping-basic.json", true));
         Analyzer analyzer = new Analyzer(
-                new Configuration(DateUtils.UTC, Protocol.FETCH_SIZE, Protocol.REQUEST_TIMEOUT,
+                new SqlConfiguration(DateUtils.UTC, Protocol.FETCH_SIZE, Protocol.REQUEST_TIMEOUT,
                                   Protocol.PAGE_TIMEOUT, null,
                                   randomFrom(Mode.values()), randomAlphaOfLength(10),
                                   null, randomAlphaOfLengthBetween(1, 15),

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/plan/logical/command/sys/SysTablesTests.java

@@ -24,7 +24,7 @@ import org.elasticsearch.xpack.sql.plan.logical.command.Command;
 import org.elasticsearch.xpack.sql.proto.Mode;
 import org.elasticsearch.xpack.sql.proto.Protocol;
 import org.elasticsearch.xpack.sql.proto.SqlTypedParamValue;
-import org.elasticsearch.xpack.sql.session.Configuration;
+import org.elasticsearch.xpack.sql.session.SqlConfiguration;
 import org.elasticsearch.xpack.sql.session.SchemaRowSet;
 import org.elasticsearch.xpack.sql.session.SqlSession;
 import org.elasticsearch.xpack.sql.stats.Metrics;
@@ -58,7 +58,7 @@ public class SysTablesTests extends ESTestCase {
     private final IndexInfo alias = new IndexInfo("alias", IndexType.ALIAS);
     private final IndexInfo frozen = new IndexInfo("frozen", IndexType.FROZEN_INDEX);
 
-    private final Configuration FROZEN_CFG = new Configuration(DateUtils.UTC, Protocol.FETCH_SIZE, Protocol.REQUEST_TIMEOUT,
+    private final SqlConfiguration FROZEN_CFG = new SqlConfiguration(DateUtils.UTC, Protocol.FETCH_SIZE, Protocol.REQUEST_TIMEOUT,
             Protocol.PAGE_TIMEOUT, null, Mode.PLAIN, null, null, null, false, true);
 
     //
@@ -331,7 +331,7 @@ public class SysTablesTests extends ESTestCase {
         return new SqlTypedParamValue(DataTypes.fromJava(value).typeName(), value);
     }
 
-    private Tuple<Command, SqlSession> sql(String sql, List<SqlTypedParamValue> params, Configuration cfg) {
+    private Tuple<Command, SqlSession> sql(String sql, List<SqlTypedParamValue> params, SqlConfiguration cfg) {
         EsIndex test = new EsIndex("test", mapping);
         Analyzer analyzer = new Analyzer(SqlTestUtils.TEST_CFG, new FunctionRegistry(), IndexResolution.valid(test),
                                          new Verifier(new Metrics()));
@@ -348,7 +348,7 @@ public class SysTablesTests extends ESTestCase {
         executeCommand(sql, emptyList(), consumer, infos);
     }
 
-    private void executeCommand(String sql, Consumer<SchemaRowSet> consumer, Configuration cfg, IndexInfo... infos) throws Exception {
+    private void executeCommand(String sql, Consumer<SchemaRowSet> consumer, SqlConfiguration cfg, IndexInfo... infos) throws Exception {
         executeCommand(sql, emptyList(), consumer, cfg, infos);
     }
 
@@ -358,7 +358,7 @@ public class SysTablesTests extends ESTestCase {
     }
 
     @SuppressWarnings({ "unchecked", "rawtypes" })
-    private void executeCommand(String sql, List<SqlTypedParamValue> params, Consumer<SchemaRowSet> consumer, Configuration cfg,
+    private void executeCommand(String sql, List<SqlTypedParamValue> params, Consumer<SchemaRowSet> consumer, SqlConfiguration cfg,
             IndexInfo... infos) throws Exception {
         Tuple<Command, SqlSession> tuple = sql(sql, params, cfg);
 

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/planner/QueryTranslatorTests.java

@@ -33,6 +33,7 @@ import org.elasticsearch.xpack.ql.querydsl.query.BoolQuery;
 import org.elasticsearch.xpack.ql.querydsl.query.ExistsQuery;
 import org.elasticsearch.xpack.ql.querydsl.query.GeoDistanceQuery;
 import org.elasticsearch.xpack.ql.querydsl.query.NotQuery;
+import org.elasticsearch.xpack.ql.querydsl.query.PrefixQuery;
 import org.elasticsearch.xpack.ql.querydsl.query.Query;
 import org.elasticsearch.xpack.ql.querydsl.query.RangeQuery;
 import org.elasticsearch.xpack.ql.querydsl.query.RegexQuery;
@@ -671,6 +672,66 @@ public class QueryTranslatorTests extends ESTestCase {
                 scriptTemplate.params().toString());
     }
 
+    public void testStartsWithUsesPrefixQuery() {
+        LogicalPlan p = plan("SELECT keyword FROM test WHERE STARTS_WITH(keyword, 'x') OR STARTS_WITH(keyword, 'y')");
+        
+        assertTrue(p instanceof Project);
+        assertTrue(p.children().get(0) instanceof Filter);
+        Expression condition = ((Filter) p.children().get(0)).condition();
+        assertFalse(condition.foldable());
+
+        QueryTranslation translation = translate(condition);
+        assertTrue(translation.query instanceof BoolQuery);
+        BoolQuery bq = (BoolQuery) translation.query;
+        
+        assertFalse(bq.isAnd());
+        assertTrue(bq.left() instanceof PrefixQuery);
+        assertTrue(bq.right() instanceof PrefixQuery);
+
+        PrefixQuery pqr = (PrefixQuery) bq.right();
+        assertEquals("keyword", pqr.field());
+        assertEquals("y", pqr.query());
+        
+        PrefixQuery pql = (PrefixQuery) bq.left();
+        assertEquals("keyword", pql.field());
+        assertEquals("x", pql.query());
+    }
+
+    public void testStartsWithUsesPrefixQueryAndScript() {
+        LogicalPlan p = plan("SELECT keyword FROM test WHERE STARTS_WITH(keyword, 'x') AND STARTS_WITH(keyword, 'xy') "
+            + "AND STARTS_WITH(LCASE(keyword), 'xyz')");
+        
+        assertTrue(p instanceof Project);
+        assertTrue(p.children().get(0) instanceof Filter);
+        Expression condition = ((Filter) p.children().get(0)).condition();
+        assertFalse(condition.foldable());
+
+        QueryTranslation translation = translate(condition);
+        assertTrue(translation.query instanceof BoolQuery);
+        BoolQuery bq = (BoolQuery) translation.query;
+        
+        assertTrue(bq.isAnd());
+        assertTrue(bq.left() instanceof BoolQuery);
+        assertTrue(bq.right() instanceof ScriptQuery);
+
+        BoolQuery bbq = (BoolQuery) bq.left();
+        assertTrue(bbq.isAnd());
+        PrefixQuery pqr = (PrefixQuery) bbq.right();
+        assertEquals("keyword", pqr.field());
+        assertEquals("xy", pqr.query());
+        
+        PrefixQuery pql = (PrefixQuery) bbq.left();
+        assertEquals("keyword", pql.field());
+        assertEquals("x", pql.query());
+
+        ScriptQuery sq = (ScriptQuery) bq.right();
+        assertEquals("InternalQlScriptUtils.nullSafeFilter(InternalQlScriptUtils.startsWith("
+                + "InternalSqlScriptUtils.lcase(InternalQlScriptUtils.docValue(doc,params.v0)), "
+                + "params.v1, params.v2))",
+            sq.script().toString());
+        assertEquals("[{v=keyword}, {v=xyz}, {v=true}]", sq.script().params().toString());
+    }
+
     public void testTranslateNotExpression_WhereClause_Painless() {
         LogicalPlan p = plan("SELECT * FROM test WHERE NOT(POSITION('x', keyword) = 0)");
         assertTrue(p instanceof Project);