SQL: Fix queries with filter resulting in NO_MATCH (#34812)

Previously, `Mapper` was returning an incorrect plan which resulted in an
```
SQLFeatureNotSupportedException: Found 1 problem(s)
line 1:8: Unexecutable item
```

Queries with a `WHERE` and/or `HAVING` clause which results in NO_MATCH
are now handled correctly and return 0 rows.

Fixes: #34613

Co-authored-by: Costin Leau <costin.leau@gmail.com>

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

@@ -64,7 +64,7 @@ protected PhysicalPlan map(LogicalPlan p) {
             }
 
             if (p instanceof LocalRelation) {
-                return new LocalExec(p.location(), (LocalRelation) p);
+                return new LocalExec(p.location(), ((LocalRelation) p).executable());
             }
 
             if (p instanceof Project) {

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

@@ -6,6 +6,7 @@
 package org.elasticsearch.xpack.sql.planner;
 
 import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
 import org.elasticsearch.xpack.sql.execution.search.AggRef;
 import org.elasticsearch.xpack.sql.expression.Alias;
 import org.elasticsearch.xpack.sql.expression.Attribute;
@@ -525,8 +526,12 @@ private static class PropagateEmptyLocal extends FoldingRule<PhysicalPlan> {
         protected PhysicalPlan rule(PhysicalPlan plan) {
             if (plan.children().size() == 1) {
                 PhysicalPlan p = plan.children().get(0);
-                if (p instanceof LocalExec && ((LocalExec) p).isEmpty()) {
-                    return new LocalExec(plan.location(), new EmptyExecutable(plan.output()));
+                if (p instanceof LocalExec) {
+                    if (((LocalExec) p).isEmpty()) {
+                        return new LocalExec(plan.location(), new EmptyExecutable(plan.output()));
+                    } else {
+                        throw new SqlIllegalArgumentException("Encountered a bug; {} is a LocalExec but is not empty", p);
+                    }
                 }
             }
             return plan;

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

@@ -0,0 +1,98 @@
+/*
+ * 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.planner;
+
+import org.elasticsearch.test.AbstractBuilderTestCase;
+import org.elasticsearch.xpack.sql.analysis.analyzer.Analyzer;
+import org.elasticsearch.xpack.sql.analysis.index.EsIndex;
+import org.elasticsearch.xpack.sql.analysis.index.IndexResolution;
+import org.elasticsearch.xpack.sql.expression.function.FunctionRegistry;
+import org.elasticsearch.xpack.sql.optimizer.Optimizer;
+import org.elasticsearch.xpack.sql.parser.SqlParser;
+import org.elasticsearch.xpack.sql.plan.physical.LocalExec;
+import org.elasticsearch.xpack.sql.plan.physical.PhysicalPlan;
+import org.elasticsearch.xpack.sql.session.EmptyExecutable;
+import org.elasticsearch.xpack.sql.type.EsField;
+import org.elasticsearch.xpack.sql.type.TypesTests;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import java.util.Map;
+import java.util.TimeZone;
+
+import static org.hamcrest.Matchers.startsWith;
+
+public class QueryFolderTests extends AbstractBuilderTestCase {
+
+    private static SqlParser parser;
+    private static Analyzer analyzer;
+    private static Optimizer optimizer;
+    private static Planner planner;
+
+    @BeforeClass
+    public static void init() {
+        parser = new SqlParser();
+
+        Map<String, EsField> mapping = TypesTests.loadMapping("mapping-multi-field-variation.json");
+        EsIndex test = new EsIndex("test", mapping);
+        IndexResolution getIndexResult = IndexResolution.valid(test);
+        analyzer = new Analyzer(new FunctionRegistry(), getIndexResult, TimeZone.getTimeZone("UTC"));
+        optimizer = new Optimizer();
+        planner = new Planner();
+    }
+
+    @AfterClass
+    public static void destroy() {
+        parser = null;
+        analyzer = null;
+    }
+
+    private PhysicalPlan plan(String sql) {
+        return planner.plan(optimizer.optimize(analyzer.analyze(parser.createStatement(sql), true)), true);
+    }
+
+    public void testFoldingToLocalExecWithProject() {
+        PhysicalPlan p = plan("SELECT keyword FROM test WHERE 1 = 2");
+        assertEquals(LocalExec.class, p.getClass());
+        LocalExec le = (LocalExec) p;
+        assertEquals(EmptyExecutable.class, le.executable().getClass());
+        EmptyExecutable ee = (EmptyExecutable) le.executable();
+        assertEquals(1, ee.output().size());
+        assertThat(ee.output().get(0).toString(), startsWith("keyword{f}#"));
+    }
+
+    public void testFoldingToLocalExecWithProject_WithOrderAndLimit() {
+        PhysicalPlan p = plan("SELECT keyword FROM test WHERE 1 = 2 ORDER BY int LIMIT 10");
+        assertEquals(LocalExec.class, p.getClass());
+        LocalExec le = (LocalExec) p;
+        assertEquals(EmptyExecutable.class, le.executable().getClass());
+        EmptyExecutable ee = (EmptyExecutable) le.executable();
+        assertEquals(1, ee.output().size());
+        assertThat(ee.output().get(0).toString(), startsWith("keyword{f}#"));
+    }
+
+    public void testFoldingToLocalExecWithProjectWithGroupBy_WithOrderAndLimit() {
+        PhysicalPlan p = plan("SELECT keyword, max(int) FROM test WHERE 1 = 2 GROUP BY keyword ORDER BY 1 LIMIT 10");
+        assertEquals(LocalExec.class, p.getClass());
+        LocalExec le = (LocalExec) p;
+        assertEquals(EmptyExecutable.class, le.executable().getClass());
+        EmptyExecutable ee = (EmptyExecutable) le.executable();
+        assertEquals(2, ee.output().size());
+        assertThat(ee.output().get(0).toString(), startsWith("keyword{f}#"));
+        assertThat(ee.output().get(1).toString(), startsWith("MAX(int){a->"));
+    }
+
+    public void testFoldingToLocalExecWithProjectWithGroupBy_WithHaving_WithOrderAndLimit() {
+        PhysicalPlan p = plan("SELECT keyword, max(int) FROM test GROUP BY keyword HAVING 1 = 2 ORDER BY 1 LIMIT 10");
+        assertEquals(LocalExec.class, p.getClass());
+        LocalExec le = (LocalExec) p;
+        assertEquals(EmptyExecutable.class, le.executable().getClass());
+        EmptyExecutable ee = (EmptyExecutable) le.executable();
+        assertEquals(2, ee.output().size());
+        assertThat(ee.output().get(0).toString(), startsWith("keyword{f}#"));
+        assertThat(ee.output().get(1).toString(), startsWith("MAX(int){a->"));
+    }
+}

x-pack/qa/sql/src/main/resources/agg.sql-spec

@@ -435,6 +435,10 @@ aggMultiGroupByMultiWithHavingUsingIn
 SELECT MIN(salary) min, MAX(salary) max, gender g, languages l, COUNT(*) c FROM "test_emp" WHERE languages > 0 GROUP BY g, languages HAVING max IN (74500, 74600) ORDER BY gender, languages;
 
 
+// HAVING filter resulting in NoMatch
+aggWithNoMatchHaving
+SELECT gender g, COUNT(*) c FROM "test_emp" GROUP BY g HAVING 1 > 2 ORDER BY gender;
+
 //
 // NULL tests
 //

x-pack/qa/sql/src/main/resources/filter.sql-spec

@@ -79,6 +79,9 @@ SELECT last_name l FROM "test_emp" WHERE emp_no BETWEEN 9990 AND 10003 ORDER BY
 whereNotBetween
 SELECT last_name l FROM "test_emp" WHERE emp_no NOT BETWEEN 10010 AND 10020 ORDER BY emp_no LIMIT 5;
 
+whereNoMatch
+SELECT last_name l FROM "test_emp" WHERE 1 = 2 ORDER BY 1 LIMIT 10;
+
 //
 // IN expression
 //