SQL: Introduce Coalesce function (#35253)

Add Coalesce conditional for replacing null values

Fix #35060

x-pack/plugin/sql/qa/security/src/test/java/org/elasticsearch/xpack/sql/qa/security/RestSqlIT.java

@@ -13,18 +13,18 @@ import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.test.rest.ESRestTestCase;
 import org.elasticsearch.xpack.sql.qa.rest.RestSqlTestCase;
 
-import static org.elasticsearch.xpack.core.security.authc.support.UsernamePasswordToken.basicAuthHeaderValue;
-
 import java.net.URISyntaxException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 
+import static org.elasticsearch.xpack.core.security.authc.support.UsernamePasswordToken.basicAuthHeaderValue;
+
 /**
  * Integration test for the rest sql action. The one that speaks json directly to a
  * user rather than to the JDBC driver or CLI.
  */
 public class RestSqlIT extends RestSqlTestCase {
-    static final boolean SSL_ENABLED = Booleans.parseBoolean(System.getProperty("tests.ssl.enabled"));
+    static final boolean SSL_ENABLED = Booleans.parseBoolean(System.getProperty("tests.ssl.enabled"), false);
 
     static Settings securitySettings() {
         String token = basicAuthHeaderValue("test_admin", new SecureString("x-pack-test-password".toCharArray()));

x-pack/plugin/sql/qa/src/main/java/org/elasticsearch/xpack/sql/qa/cli/ShowTestCase.java

@@ -35,6 +35,10 @@ public abstract class ShowTestCase extends CliIntegrationTestCase {
         while (aggregateFunction.matcher(line).matches()) {
             line = readLine();
         }
+        Pattern conditionalFunction = Pattern.compile("\\s*[A-Z0-9_~]+\\s*\\|\\s*CONDITIONAL\\s*");
+        while (conditionalFunction.matcher(line).matches()) {
+            line = readLine();
+        }
         Pattern scalarFunction = Pattern.compile("\\s*[A-Z0-9_~]+\\s*\\|\\s*SCALAR\\s*");
         while (scalarFunction.matcher(line).matches()) {
             line = readLine();

x-pack/plugin/sql/qa/src/main/java/org/elasticsearch/xpack/sql/qa/jdbc/CsvSpecTestCase.java

@@ -36,7 +36,7 @@ public abstract class CsvSpecTestCase extends SpecBaseIntegrationTestCase {
         tests.addAll(readScriptSpec("/columns.csv-spec", parser));
         tests.addAll(readScriptSpec("/datetime.csv-spec", parser));
         tests.addAll(readScriptSpec("/alias.csv-spec", parser));
-        tests.addAll(readScriptSpec("/nulls.csv-spec", parser));
+        tests.addAll(readScriptSpec("/null.csv-spec", parser));
         tests.addAll(readScriptSpec("/nested.csv-spec", parser));
         tests.addAll(readScriptSpec("/functions.csv-spec", parser));
         tests.addAll(readScriptSpec("/math.csv-spec", parser));

x-pack/plugin/sql/qa/src/main/java/org/elasticsearch/xpack/sql/qa/jdbc/JdbcAssert.java

@@ -114,6 +114,11 @@ public class JdbcAssert {
             if (expectedType == Types.FLOAT && expected instanceof CsvResultSet) {
                 expectedType = Types.REAL;
             }
+            // csv doesn't support NULL type so skip type checking
+            if (actualType == Types.NULL && expected instanceof CsvResultSet) {
+                expectedType = Types.NULL;
+            }
+
             // when lenient is used, an int is equivalent to a short, etc...
             assertEquals("Different column type for column [" + expectedName + "] (" + JDBCType.valueOf(expectedType) + " != "
                     + JDBCType.valueOf(actualType) + ")", expectedType, actualType);

x-pack/plugin/sql/qa/src/main/java/org/elasticsearch/xpack/sql/qa/jdbc/SqlSpecTestCase.java

@@ -41,6 +41,7 @@ public abstract class SqlSpecTestCase extends SpecBaseIntegrationTestCase {
         tests.addAll(readScriptSpec("/arithmetic.sql-spec", parser));
         tests.addAll(readScriptSpec("/string-functions.sql-spec", parser));
         tests.addAll(readScriptSpec("/case-functions.sql-spec", parser));
+        tests.addAll(readScriptSpec("/null.sql-spec", parser));
         return tests;
     }
 

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

@@ -19,6 +19,7 @@ SKEWNESS        |AGGREGATE
 STDDEV_POP      |AGGREGATE      
 SUM_OF_SQUARES  |AGGREGATE      
 VAR_POP         |AGGREGATE      
+COALESCE        |CONDITIONAL
 DAY             |SCALAR         
 DAYNAME         |SCALAR         
 DAYOFMONTH      |SCALAR         

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

@@ -196,6 +196,7 @@ SKEWNESS        |AGGREGATE
 STDDEV_POP      |AGGREGATE      
 SUM_OF_SQUARES  |AGGREGATE      
 VAR_POP         |AGGREGATE      
+COALESCE        |CONDITIONAL
 DAY             |SCALAR         
 DAYNAME         |SCALAR         
 DAYOFMONTH      |SCALAR         

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

@@ -0,0 +1,75 @@
+//
+// Null expressions
+//
+
+dateTimeOverNull
+SELECT YEAR(CAST(NULL AS DATE)) d;
+
+d:i
+null
+;
+
+addOfNull
+SELECT CAST(NULL AS INT) + CAST(NULL AS FLOAT) AS n;
+
+n:d
+null
+;
+
+
+divOfCastedNull
+SELECT 5 / CAST(NULL AS FLOAT) + 10 AS n;
+
+n:d
+null
+;
+
+divNoNull
+SELECT 5 / null + 1 AS n;
+
+n:i
+null
+;
+
+coalesceJustWithNull
+SELECT COALESCE(null, null, null) AS c;
+
+c
+null
+;
+
+coalesceFirstNotNull
+SELECT COALESCE(123) AS c;
+
+c
+123
+;
+
+
+coalesceWithFirstNullOfString
+SELECT COALESCE(null, 'first') AS c;
+
+c:s
+first
+;
+
+coalesceWithFirstNullOfNumber
+SELECT COALESCE(null, 123) AS c;
+
+c:i
+123
+;
+
+coalesceMixed
+SELECT COALESCE(null, 123, null, 321) AS c;
+
+c:i
+123
+;
+
+coalesceScalar
+SELECT COALESCE(null, ABS(123) + 1) AS c;
+
+c:i
+124
+;

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

@@ -0,0 +1,12 @@
+//
+// Null expressions
+//
+
+coalesceField
+SELECT COALESCE(null, ABS(emp_no) + 1) AS c FROM test_emp ORDER BY emp_no LIMIT 5;
+
+coalesceHaving
+SELECT COALESCE(null, ABS(MAX(emp_no)) + 1, 123) AS c FROM test_emp GROUP BY languages HAVING c > 100 ORDER BY languages LIMIT 5;
+
+coalesceWhere
+SELECT COALESCE(null, ABS(emp_no) + 1, 123) AS c FROM test_emp WHERE COALESCE(null, ABS(emp_no) + 1, 123, 321) > 100 ORDER BY emp_no NULLS FIRST LIMIT 5;

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

@@ -1,25 +0,0 @@
-//
-// Null expressions
-//
-
-nullDate
-SELECT YEAR(CAST(NULL AS DATE)) AS d;
-
-d:i
-null
-;
-
-nullAdd
-SELECT CAST(NULL AS INT) + CAST(NULL AS FLOAT) AS n;
-
-n:d
-null
-;
-
-
-nullDiv
-SELECT 5 / CAST(NULL AS FLOAT) + 10 AS n;
-
-n:d
-null
-;

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

@@ -18,6 +18,7 @@ import java.util.List;
 import java.util.Locale;
 import java.util.function.Predicate;
 
+import static java.lang.String.format;
 import static java.util.Collections.emptyList;
 import static java.util.Collections.emptyMap;
 
@@ -103,6 +104,10 @@ public final class Expressions {
         return e instanceof NamedExpression ? ((NamedExpression) e).name() : e.nodeName();
     }
 
+    public static boolean isNull(Expression e) {
+        return e.dataType() == DataType.NULL || (e.foldable() && e.fold() == null);
+    }
+
     public static List<String> names(Collection<? extends Expression> e) {
         List<String> names = new ArrayList<>(e.size());
         for (Expression ex : e) {
@@ -137,6 +142,14 @@ public final class Expressions {
         throw new SqlIllegalArgumentException("Cannot create pipe for {}", e);
     }
 
+    public static List<Pipe> pipe(List<Expression> expressions) {
+        List<Pipe> pipes = new ArrayList<>(expressions.size());
+        for (Expression e : expressions) {
+            pipes.add(pipe(e));
+        }
+        return pipes;
+    }
+
     public static TypeResolution typeMustBeBoolean(Expression e, String operationName, ParamOrdinal paramOrd) {
         return typeMustBe(e, dt -> dt == DataType.BOOLEAN, operationName, paramOrd, "boolean");
     }
@@ -161,27 +174,18 @@ public final class Expressions {
         return typeMustBe(e, dt -> dt.isNumeric() || dt == DataType.DATE, operationName, paramOrd, "numeric", "date");
     }
 
-    private static TypeResolution typeMustBe(Expression e,
+    public static TypeResolution typeMustBe(Expression e,
                                              Predicate<DataType> predicate,
-                                             String operationName,
-                                             ParamOrdinal pOrd,
-                                             String... acceptedTypes) {
-
-        return predicate.test(e.dataType()) || DataTypes.isNull(e.dataType())?
-            TypeResolution.TYPE_RESOLVED :
-            new TypeResolution(incorrectTypeErrorMessage(e, operationName, pOrd, acceptedTypes));
-
-    }
-
-    private static String incorrectTypeErrorMessage(Expression e,
                                              String operationName,
                                              ParamOrdinal paramOrd,
                                              String... acceptedTypes) {
-        return String.format(Locale.ROOT, "[%s]%s argument must be [%s], found value [%s] type [%s]",
+        return predicate.test(e.dataType()) || DataTypes.isNull(e.dataType())?
+            TypeResolution.TYPE_RESOLVED :
+            new TypeResolution(format(Locale.ROOT, "[%s]%s argument must be [%s], found value [%s] type [%s]",
                     operationName,
                     paramOrd == null || paramOrd == ParamOrdinal.DEFAULT ? "" : " " + paramOrd.name().toLowerCase(Locale.ROOT),
                     Strings.arrayToDelimitedString(acceptedTypes, " or "),
                     Expressions.name(e),
-            e.dataType().esType);
+                    e.dataType().esType));
     }
 }

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

@@ -82,6 +82,7 @@ import org.elasticsearch.xpack.sql.expression.function.scalar.string.Right;
 import org.elasticsearch.xpack.sql.expression.function.scalar.string.Space;
 import org.elasticsearch.xpack.sql.expression.function.scalar.string.Substring;
 import org.elasticsearch.xpack.sql.expression.function.scalar.string.UCase;
+import org.elasticsearch.xpack.sql.expression.predicate.conditional.Coalesce;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.arithmetic.Mod;
 import org.elasticsearch.xpack.sql.parser.ParsingException;
 import org.elasticsearch.xpack.sql.tree.Location;
@@ -142,6 +143,8 @@ public class FunctionRegistry {
                 def(Skewness.class, Skewness::new),
                 def(Kurtosis.class, Kurtosis::new));
         // Scalar functions
+        // conditional
+        addToMap(def(Coalesce.class, Coalesce::new));
         // Date
         addToMap(def(DayName.class, DayName::new, "DAYNAME"),
                 def(DayOfMonth.class, DayOfMonth::new, "DAYOFMONTH", "DAY", "DOM"),
@@ -310,6 +313,26 @@ public class FunctionRegistry {
         return def(function, builder, false, aliases);
     }
 
+    /**
+     * Build a {@linkplain FunctionDefinition} for multi-arg function that
+     * is not aware of time zone and does not support {@code DISTINCT}.
+     */
+    @SuppressWarnings("overloads") // These are ambiguous if you aren't using ctor references but we always do
+    static <T extends Function> FunctionDefinition def(Class<T> function,
+            MultiFunctionBuilder<T> ctorRef, String... aliases) {
+        FunctionBuilder builder = (location, children, distinct, tz) -> {
+            if (distinct) {
+                throw new IllegalArgumentException("does not support DISTINCT yet it was specified");
+            }
+            return ctorRef.build(location, children);
+        };
+        return def(function, builder, false, aliases);
+    }
+
+    interface MultiFunctionBuilder<T> {
+        T build(Location location, List<Expression> children);
+    }
+    
     /**
      * Build a {@linkplain FunctionDefinition} for a unary function that is not
      * aware of time zone but does support {@code DISTINCT}.
@@ -325,6 +348,7 @@ public class FunctionRegistry {
         };
         return def(function, builder, false, aliases);
     }
+
     interface DistinctAwareUnaryFunctionBuilder<T> {
         T build(Location location, Expression target, boolean distinct);
     }
@@ -347,6 +371,7 @@ public class FunctionRegistry {
         };
         return def(function, builder, true, aliases);
     }
+
     interface DatetimeUnaryFunctionBuilder<T> {
         T build(Location location, Expression target, TimeZone tz);
     }
@@ -373,6 +398,7 @@ public class FunctionRegistry {
         };
         return def(function, builder, false, aliases);
     }
+
     interface BinaryFunctionBuilder<T> {
         T build(Location location, Expression lhs, Expression rhs);
     }
@@ -391,6 +417,7 @@ public class FunctionRegistry {
         };
         return new FunctionDefinition(primaryName, unmodifiableList(Arrays.asList(aliases)), function, datetime, realBuilder);
     }
+
     private interface FunctionBuilder {
         Function build(Location location, List<Expression> children, boolean distinct, TimeZone tz);
     }
@@ -450,6 +477,7 @@ public class FunctionRegistry {
             ctorRef.build(location, children.get(0), children.get(0).dataType());
         return def(function, builder, false, aliases);
     }
+
     private interface CastFunctionBuilder<T> {
         T build(Location location, Expression expression, DataType dataType);
     }

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

@@ -8,11 +8,13 @@ package org.elasticsearch.xpack.sql.expression.function;
 import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
 import org.elasticsearch.xpack.sql.expression.function.aggregate.AggregateFunction;
 import org.elasticsearch.xpack.sql.expression.function.scalar.ScalarFunction;
+import org.elasticsearch.xpack.sql.expression.predicate.conditional.ConditionalFunction;
 
 
 public enum FunctionType {
 
     AGGREGATE(AggregateFunction.class),
+    CONDITIONAL(ConditionalFunction.class),
     SCALAR(ScalarFunction.class),
     SCORE(Score.class);
 

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

@@ -25,9 +25,10 @@ import org.elasticsearch.xpack.sql.expression.gen.processor.ChainingProcessor;
 import org.elasticsearch.xpack.sql.expression.gen.processor.ConstantProcessor;
 import org.elasticsearch.xpack.sql.expression.gen.processor.HitExtractorProcessor;
 import org.elasticsearch.xpack.sql.expression.gen.processor.Processor;
-import org.elasticsearch.xpack.sql.expression.predicate.IsNotNullProcessor;
+import org.elasticsearch.xpack.sql.expression.predicate.conditional.CoalesceProcessor;
 import org.elasticsearch.xpack.sql.expression.predicate.logical.BinaryLogicProcessor;
 import org.elasticsearch.xpack.sql.expression.predicate.logical.NotProcessor;
+import org.elasticsearch.xpack.sql.expression.predicate.nulls.IsNotNullProcessor;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.arithmetic.BinaryArithmeticProcessor;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.arithmetic.UnaryArithmeticProcessor;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.comparison.BinaryComparisonProcessor;
@@ -58,6 +59,7 @@ public final class Processors {
         entries.add(new Entry(Processor.class, BinaryLogicProcessor.NAME, BinaryLogicProcessor::new));
         entries.add(new Entry(Processor.class, NotProcessor.NAME, NotProcessor::new));
         // null
+        entries.add(new Entry(Processor.class, CoalesceProcessor.NAME, CoalesceProcessor::new));
         entries.add(new Entry(Processor.class, IsNotNullProcessor.NAME, IsNotNullProcessor::new));
 
         // arithmetic

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

@@ -21,9 +21,10 @@ import org.elasticsearch.xpack.sql.expression.function.scalar.string.LocateFunct
 import org.elasticsearch.xpack.sql.expression.function.scalar.string.ReplaceFunctionProcessor;
 import org.elasticsearch.xpack.sql.expression.function.scalar.string.StringProcessor.StringOperation;
 import org.elasticsearch.xpack.sql.expression.function.scalar.string.SubstringFunctionProcessor;
-import org.elasticsearch.xpack.sql.expression.predicate.IsNotNullProcessor;
 import org.elasticsearch.xpack.sql.expression.predicate.logical.BinaryLogicProcessor.BinaryLogicOperation;
+import org.elasticsearch.xpack.sql.expression.predicate.conditional.CoalesceProcessor;
 import org.elasticsearch.xpack.sql.expression.predicate.logical.NotProcessor;
+import org.elasticsearch.xpack.sql.expression.predicate.nulls.IsNotNullProcessor;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.arithmetic.BinaryArithmeticProcessor.BinaryArithmeticOperation;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.arithmetic.UnaryArithmeticProcessor.UnaryArithmeticOperation;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.comparison.BinaryComparisonProcessor.BinaryComparisonOperation;
@@ -124,6 +125,13 @@ public final class InternalSqlScriptUtils {
         return InProcessor.apply(value, values);
     }
 
+    //
+    // Null
+    //
+    public static Object coalesce(List<Object> expressions) {
+        return CoalesceProcessor.apply(expressions);
+    }
+
     //
     // Regex
     //

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/gen/pipeline/MultiPipe.java

@@ -0,0 +1,33 @@
+/*
+ * 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.sql.expression.gen.pipeline;
+
+import org.elasticsearch.xpack.sql.expression.Expression;
+import org.elasticsearch.xpack.sql.expression.gen.processor.Processor;
+import org.elasticsearch.xpack.sql.tree.Location;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public abstract class MultiPipe extends Pipe {
+
+    protected MultiPipe(Location location, Expression expression, List<Pipe> children) {
+        super(location, expression, children);
+    }
+
+    @Override
+    public Processor asProcessor() {
+        List<Processor> procs = new ArrayList<>();
+        for (Pipe pipe : children()) {
+            procs.add(pipe.asProcessor());
+        }
+
+        return asProcessor(procs);
+    }
+
+    public abstract Processor asProcessor(List<Processor> procs);
+}

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/gen/pipeline/Pipe.java

@@ -6,13 +6,16 @@
 package org.elasticsearch.xpack.sql.expression.gen.pipeline;
 
 import org.elasticsearch.xpack.sql.capabilities.Resolvable;
+import org.elasticsearch.xpack.sql.capabilities.Resolvables;
 import org.elasticsearch.xpack.sql.execution.search.FieldExtraction;
+import org.elasticsearch.xpack.sql.execution.search.SqlSourceBuilder;
 import org.elasticsearch.xpack.sql.expression.Attribute;
 import org.elasticsearch.xpack.sql.expression.Expression;
 import org.elasticsearch.xpack.sql.expression.gen.processor.Processor;
 import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.tree.Node;
 
+import java.util.ArrayList;
 import java.util.List;
 
 /**
@@ -38,6 +41,27 @@ public abstract class Pipe extends Node<Pipe> implements FieldExtraction, Resolv
         return expression;
     }
 
+    @Override
+    public boolean resolved() {
+        return Resolvables.resolved(children());
+    }
+
+    @Override
+    public void collectFields(SqlSourceBuilder sourceBuilder) {
+        children().forEach(c -> c.collectFields(sourceBuilder));
+    }
+
+    @Override
+    public boolean supportedByAggsOnlyQuery() {
+        for (Pipe pipe : children()) {
+            if (pipe.supportedByAggsOnlyQuery()) {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
     public abstract Processor asProcessor();
 
     /**
@@ -47,7 +71,14 @@ public abstract class Pipe extends Node<Pipe> implements FieldExtraction, Resolv
      * @return {@code this} if the resolution doesn't change the
      *      definition, a new {@link Pipe} otherwise
      */
-    public abstract Pipe resolveAttributes(AttributeResolver resolver);
+    public Pipe resolveAttributes(AttributeResolver resolver) {
+        List<Pipe> newPipes = new ArrayList<>(children().size());
+        for (Pipe p : children()) {
+            newPipes.add(p.resolveAttributes(resolver));
+        }
+
+        return children().equals(newPipes) ? this : replaceChildren(newPipes);
+    }
 
     public interface AttributeResolver {
         FieldExtraction resolve(Attribute attribute);

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/predicate/conditional/Coalesce.java

@@ -0,0 +1,93 @@
+/*
+ * 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.sql.expression.predicate.conditional;
+
+import org.elasticsearch.xpack.sql.expression.Expression;
+import org.elasticsearch.xpack.sql.expression.Expressions;
+import org.elasticsearch.xpack.sql.expression.gen.pipeline.Pipe;
+import org.elasticsearch.xpack.sql.expression.gen.script.ParamsBuilder;
+import org.elasticsearch.xpack.sql.expression.gen.script.ScriptTemplate;
+import org.elasticsearch.xpack.sql.tree.Location;
+import org.elasticsearch.xpack.sql.tree.NodeInfo;
+import org.elasticsearch.xpack.sql.type.DataType;
+import org.elasticsearch.xpack.sql.type.DataTypeConversion;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.StringJoiner;
+
+import static org.elasticsearch.xpack.sql.expression.gen.script.ParamsBuilder.paramsBuilder;
+
+public class Coalesce extends ConditionalFunction {
+
+    private DataType dataType = DataType.NULL;
+
+    public Coalesce(Location location, List<Expression> fields) {
+        super(location, fields);
+    }
+
+    @Override
+    protected NodeInfo<Coalesce> info() {
+        return NodeInfo.create(this, Coalesce::new, children());
+    }
+
+    @Override
+    public Expression replaceChildren(List<Expression> newChildren) {
+        return new Coalesce(location(), newChildren);
+    }
+
+    @Override
+    protected TypeResolution resolveType() {
+        for (Expression e : children()) {
+            dataType = DataTypeConversion.commonType(dataType, e.dataType());
+        }
+        return TypeResolution.TYPE_RESOLVED;
+    }
+
+    @Override
+    public DataType dataType() {
+        return dataType;
+    }
+
+    @Override
+    public boolean foldable() {
+        // if the first entry is foldable, so is coalesce
+        // that's because the nulls are eliminated by the optimizer
+        // and if the first expression is folded (and not null), the rest do not matter
+        List<Expression> children = children();
+        return (children.isEmpty() || (children.get(0).foldable() && children.get(0).fold() != null));
+    }
+
+    @Override
+    public Object fold() {
+        List<Expression> children = children();
+        return children.isEmpty() ? null : children.get(0).fold();
+    }
+
+    @Override
+    public ScriptTemplate asScript() {
+        List<ScriptTemplate> templates = new ArrayList<>();
+        for (Expression ex : children()) {
+            templates.add(asScript(ex));
+        }
+
+        StringJoiner template = new StringJoiner(",", "{sql}.coalesce([", "])");
+        ParamsBuilder params = paramsBuilder();
+
+        for (ScriptTemplate scriptTemplate : templates) {
+            template.add(scriptTemplate.template());
+            params.script(scriptTemplate.params());
+        }
+
+        return new ScriptTemplate(template.toString(), params.build(), dataType);
+    }
+
+    @Override
+    protected Pipe makePipe() {
+        return new CoalescePipe(location(), this, Expressions.pipe(children()));
+    }
+}

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/predicate/conditional/CoalescePipe.java

@@ -0,0 +1,38 @@
+/*
+ * 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.sql.expression.predicate.conditional;
+
+import org.elasticsearch.xpack.sql.expression.Expression;
+import org.elasticsearch.xpack.sql.expression.gen.pipeline.MultiPipe;
+import org.elasticsearch.xpack.sql.expression.gen.pipeline.Pipe;
+import org.elasticsearch.xpack.sql.expression.gen.processor.Processor;
+import org.elasticsearch.xpack.sql.tree.Location;
+import org.elasticsearch.xpack.sql.tree.NodeInfo;
+
+import java.util.List;
+
+public class CoalescePipe extends MultiPipe {
+
+    public CoalescePipe(Location location, Expression expression, List<Pipe> children) {
+        super(location, expression, children);
+    }
+
+    @Override
+    protected NodeInfo<CoalescePipe> info() {
+        return NodeInfo.create(this, CoalescePipe::new, expression(), children());
+    }
+
+    @Override
+    public Pipe replaceChildren(List<Pipe> newChildren) {
+        return new CoalescePipe(location(), expression(), newChildren);
+    }
+
+    @Override
+    public Processor asProcessor(List<Processor> procs) {
+        return new CoalesceProcessor(procs);
+    }
+}

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/predicate/conditional/CoalesceProcessor.java

@@ -0,0 +1,81 @@
+/*
+ * 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.sql.expression.predicate.conditional;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.xpack.sql.expression.gen.processor.Processor;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+public class CoalesceProcessor implements Processor {
+    
+    public static final String NAME = "nco";
+
+    private final List<Processor> processsors;
+
+    public CoalesceProcessor(List<Processor> processors) {
+        this.processsors = processors;
+    }
+
+    public CoalesceProcessor(StreamInput in) throws IOException {
+        processsors = in.readNamedWriteableList(Processor.class);
+    }
+
+    @Override
+    public String getWriteableName() {
+        return NAME;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeNamedWriteableList(processsors);
+    }
+
+    @Override
+    public Object process(Object input) {
+        for (Processor proc : processsors) {
+            Object result = proc.process(input);
+            if (result != null) {
+                return result;
+            }
+        }
+        return null;
+    }
+
+    public static Object apply(List<?> values) {
+        if (values == null || values.isEmpty()) {
+            return null;
+        }
+
+        for (Object object : values) {
+            if (object != null) {
+                return object;
+            }
+        }
+
+        return null;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(processsors);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        CoalesceProcessor that = (CoalesceProcessor) o;
+        return Objects.equals(processsors, that.processsors);
+    }
+}

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/predicate/conditional/ConditionalFunction.java

@@ -0,0 +1,23 @@
+/*
+ * 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.sql.expression.predicate.conditional;
+
+import org.elasticsearch.xpack.sql.expression.Expression;
+import org.elasticsearch.xpack.sql.expression.function.scalar.ScalarFunction;
+import org.elasticsearch.xpack.sql.tree.Location;
+
+import java.util.List;
+
+/**
+ * Base class for conditional predicates.
+ */
+public abstract class ConditionalFunction extends ScalarFunction {
+
+    protected ConditionalFunction(Location location, List<Expression> fields) {
+        super(location, fields);
+    }
+}

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/predicate/logical/NotProcessor.java

@@ -14,7 +14,7 @@ import java.io.IOException;
 
 public class NotProcessor implements Processor {
     
-    static final NotProcessor INSTANCE = new NotProcessor();
+    public static final NotProcessor INSTANCE = new NotProcessor();
 
     public static final String NAME = "ln";
 

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/predicate/nulls/IsNotNull.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.sql.expression.predicate;
+package org.elasticsearch.xpack.sql.expression.predicate.nulls;
 
 import org.elasticsearch.xpack.sql.expression.Expression;
 import org.elasticsearch.xpack.sql.expression.function.scalar.UnaryScalarFunction;

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/predicate/nulls/IsNotNullProcessor.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.sql.expression.predicate;
+package org.elasticsearch.xpack.sql.expression.predicate.nulls;
 
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -15,7 +15,7 @@ public class IsNotNullProcessor implements Processor {
     
     static final IsNotNullProcessor INSTANCE = new IsNotNullProcessor();
 
-    public static final String NAME = "inn";
+    public static final String NAME = "ninn";
 
     private IsNotNullProcessor() {}
 

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

@@ -39,12 +39,13 @@ import org.elasticsearch.xpack.sql.expression.function.scalar.ScalarFunctionAttr
 import org.elasticsearch.xpack.sql.expression.predicate.BinaryOperator;
 import org.elasticsearch.xpack.sql.expression.predicate.BinaryOperator.Negateable;
 import org.elasticsearch.xpack.sql.expression.predicate.BinaryPredicate;
-import org.elasticsearch.xpack.sql.expression.predicate.IsNotNull;
 import org.elasticsearch.xpack.sql.expression.predicate.Predicates;
 import org.elasticsearch.xpack.sql.expression.predicate.Range;
+import org.elasticsearch.xpack.sql.expression.predicate.conditional.Coalesce;
 import org.elasticsearch.xpack.sql.expression.predicate.logical.And;
 import org.elasticsearch.xpack.sql.expression.predicate.logical.Not;
 import org.elasticsearch.xpack.sql.expression.predicate.logical.Or;
+import org.elasticsearch.xpack.sql.expression.predicate.nulls.IsNotNull;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.comparison.BinaryComparison;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.comparison.Equals;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.comparison.GreaterThan;
@@ -128,6 +129,7 @@ public class Optimizer extends RuleExecutor<LogicalPlan> {
                 new ReplaceFoldableAttributes(),
                 new FoldNull(),
                 new ConstantFolding(),
+                new SimplifyCoalesce(),
                 // boolean
                 new BooleanSimplification(),
                 new BooleanLiteralsOnTheRight(),
@@ -687,7 +689,7 @@ public class Optimizer extends RuleExecutor<LogicalPlan> {
                 if (TRUE.equals(filter.condition())) {
                     return filter.child();
                 }
-                if (FALSE.equals(filter.condition()) || FoldNull.isNull(filter.condition())) {
+                if (FALSE.equals(filter.condition()) || Expressions.isNull(filter.condition())) {
                     return new LocalRelation(filter.location(), new EmptyExecutable(filter.output()));
                 }
             }
@@ -1122,10 +1124,6 @@ public class Optimizer extends RuleExecutor<LogicalPlan> {
             super(TransformDirection.UP);
         }
 
-        private static boolean isNull(Expression ex) {
-            return DataType.NULL == ex.dataType() || (ex.foldable() && ex.fold() == null);
-        }
-
         @Override
         protected Expression rule(Expression e) {
             if (e instanceof IsNotNull) {
@@ -1138,12 +1136,12 @@ public class Optimizer extends RuleExecutor<LogicalPlan> {
 
             if (e instanceof In) {
                 In in = (In) e;
-                if (isNull(in.value())) {
+                if (Expressions.isNull(in.value())) {
                     return Literal.of(in, null);
                 }
             }
 
-            if (e.nullable() && Expressions.anyMatch(e.children(), FoldNull::isNull)) {
+            if (e.nullable() && Expressions.anyMatch(e.children(), Expressions::isNull)) {
                 return Literal.of(e, null);
             }
 
@@ -1168,6 +1166,38 @@ public class Optimizer extends RuleExecutor<LogicalPlan> {
         }
     }
     
+    static class SimplifyCoalesce extends OptimizerExpressionRule {
+
+        SimplifyCoalesce() {
+            super(TransformDirection.DOWN);
+        }
+
+        @Override
+        protected Expression rule(Expression e) {
+            if (e instanceof Coalesce) {
+                Coalesce c = (Coalesce) e;
+
+                // find the first non-null foldable child (if any) and remove the rest
+                // while at it, exclude any nulls found
+                List<Expression> newChildren = new ArrayList<>();
+
+                for (Expression child : c.children()) {
+                    if (Expressions.isNull(child) == false) {
+                        newChildren.add(child);
+                        if (child.foldable()) {
+                            break;
+                        }
+                    }
+                }
+
+                if (newChildren.size() < c.children().size()) {
+                    return new Coalesce(c.location(), newChildren);
+                }
+            }
+            return e;
+        }
+    }
+
     static class BooleanSimplification extends OptimizerExpressionRule {
 
         BooleanSimplification() {

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

@@ -25,7 +25,6 @@ import org.elasticsearch.xpack.sql.expression.function.Function;
 import org.elasticsearch.xpack.sql.expression.function.UnresolvedFunction;
 import org.elasticsearch.xpack.sql.expression.function.scalar.Cast;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.comparison.In;
-import org.elasticsearch.xpack.sql.expression.predicate.IsNotNull;
 import org.elasticsearch.xpack.sql.expression.predicate.Range;
 import org.elasticsearch.xpack.sql.expression.predicate.fulltext.MatchQueryPredicate;
 import org.elasticsearch.xpack.sql.expression.predicate.fulltext.MultiMatchQueryPredicate;
@@ -33,6 +32,7 @@ import org.elasticsearch.xpack.sql.expression.predicate.fulltext.StringQueryPred
 import org.elasticsearch.xpack.sql.expression.predicate.logical.And;
 import org.elasticsearch.xpack.sql.expression.predicate.logical.Not;
 import org.elasticsearch.xpack.sql.expression.predicate.logical.Or;
+import org.elasticsearch.xpack.sql.expression.predicate.nulls.IsNotNull;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.arithmetic.Add;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.arithmetic.Div;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.arithmetic.Mod;

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

@@ -31,7 +31,6 @@ import org.elasticsearch.xpack.sql.expression.function.scalar.ScalarFunction;
 import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeFunction;
 import org.elasticsearch.xpack.sql.expression.function.scalar.datetime.DateTimeHistogramFunction;
 import org.elasticsearch.xpack.sql.expression.gen.script.ScriptTemplate;
-import org.elasticsearch.xpack.sql.expression.predicate.IsNotNull;
 import org.elasticsearch.xpack.sql.expression.predicate.Range;
 import org.elasticsearch.xpack.sql.expression.predicate.fulltext.MatchQueryPredicate;
 import org.elasticsearch.xpack.sql.expression.predicate.fulltext.MultiMatchQueryPredicate;
@@ -39,6 +38,7 @@ import org.elasticsearch.xpack.sql.expression.predicate.fulltext.StringQueryPred
 import org.elasticsearch.xpack.sql.expression.predicate.logical.And;
 import org.elasticsearch.xpack.sql.expression.predicate.logical.Not;
 import org.elasticsearch.xpack.sql.expression.predicate.logical.Or;
+import org.elasticsearch.xpack.sql.expression.predicate.nulls.IsNotNull;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.comparison.BinaryComparison;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.comparison.Equals;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.comparison.GreaterThan;

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

@@ -102,7 +102,7 @@ public abstract class DataTypeConversion {
         if (from == to) {
             return Conversion.IDENTITY;
         }
-        if (to == DataType.NULL) {
+        if (to == DataType.NULL || from == DataType.NULL) {
             return Conversion.NULL;
         }
         if (from == DataType.NULL) {

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

@@ -35,6 +35,11 @@ class org.elasticsearch.xpack.sql.expression.function.scalar.whitelist.InternalS
   Boolean not(Boolean)
   Boolean notNull(Object)    
 
+#
+# Null
+#
+  Object coalesce(java.util.List)
+
 #
 # Regex
 #

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

@@ -35,11 +35,12 @@ import org.elasticsearch.xpack.sql.expression.function.scalar.math.Floor;
 import org.elasticsearch.xpack.sql.expression.function.scalar.string.Ascii;
 import org.elasticsearch.xpack.sql.expression.function.scalar.string.Repeat;
 import org.elasticsearch.xpack.sql.expression.predicate.BinaryOperator;
-import org.elasticsearch.xpack.sql.expression.predicate.IsNotNull;
 import org.elasticsearch.xpack.sql.expression.predicate.Range;
+import org.elasticsearch.xpack.sql.expression.predicate.conditional.Coalesce;
 import org.elasticsearch.xpack.sql.expression.predicate.logical.And;
 import org.elasticsearch.xpack.sql.expression.predicate.logical.Not;
 import org.elasticsearch.xpack.sql.expression.predicate.logical.Or;
+import org.elasticsearch.xpack.sql.expression.predicate.nulls.IsNotNull;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.arithmetic.Add;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.arithmetic.Div;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.arithmetic.Mod;
@@ -66,6 +67,7 @@ import org.elasticsearch.xpack.sql.optimizer.Optimizer.PropagateEquals;
 import org.elasticsearch.xpack.sql.optimizer.Optimizer.PruneDuplicateFunctions;
 import org.elasticsearch.xpack.sql.optimizer.Optimizer.PruneSubqueryAliases;
 import org.elasticsearch.xpack.sql.optimizer.Optimizer.ReplaceFoldableAttributes;
+import org.elasticsearch.xpack.sql.optimizer.Optimizer.SimplifyCoalesce;
 import org.elasticsearch.xpack.sql.plan.logical.Filter;
 import org.elasticsearch.xpack.sql.plan.logical.LocalRelation;
 import org.elasticsearch.xpack.sql.plan.logical.LogicalPlan;
@@ -78,6 +80,7 @@ import org.elasticsearch.xpack.sql.tree.Location;
 import org.elasticsearch.xpack.sql.tree.NodeInfo;
 import org.elasticsearch.xpack.sql.type.DataType;
 import org.elasticsearch.xpack.sql.type.EsField;
+import org.elasticsearch.xpack.sql.util.CollectionUtils;
 
 import java.util.Arrays;
 import java.util.Collections;
@@ -402,6 +405,40 @@ public class OptimizerTests extends ESTestCase {
         assertNullLiteral(rule.rule(new RLike(EMPTY, getFieldAttribute(), Literal.NULL)));
     }
 
+    public void testSimplifyCoalesceNulls() {
+        Expression e = new SimplifyCoalesce().rule(new Coalesce(EMPTY, asList(Literal.NULL, Literal.NULL)));
+        assertEquals(Coalesce.class, e.getClass());
+        assertEquals(0, e.children().size());
+    }
+
+    public void testSimplifyCoalesceRandomNulls() {
+        Expression e = new SimplifyCoalesce().rule(new Coalesce(EMPTY, randomListOfNulls()));
+        assertEquals(Coalesce.class, e.getClass());
+        assertEquals(0, e.children().size());
+    }
+
+    public void testSimplifyCoalesceRandomNullsWithValue() {
+        Expression e = new SimplifyCoalesce().rule(new Coalesce(EMPTY,
+                CollectionUtils.combine(
+                        CollectionUtils.combine(randomListOfNulls(), Literal.TRUE, Literal.FALSE, Literal.TRUE),
+                        randomListOfNulls())));
+        assertEquals(1, e.children().size());
+        assertEquals(Literal.TRUE, e.children().get(0));
+    }
+
+    private List<Expression> randomListOfNulls() {
+        return asList(randomArray(1, 10, i -> new Literal[i], () -> Literal.NULL));
+    }
+
+    public void testSimplifyCoalesceFirstLiteral() {
+        Expression e = new SimplifyCoalesce()
+                .rule(new Coalesce(EMPTY,
+                        Arrays.asList(Literal.NULL, Literal.TRUE, Literal.FALSE, new Abs(EMPTY, getFieldAttribute()))));
+        assertEquals(Coalesce.class, e.getClass());
+        assertEquals(1, e.children().size());
+        assertEquals(Literal.TRUE, e.children().get(0));
+    }
+
     //
     // Logical simplifications
     //