ESQL: Support StubRelation with empty output in PropagateEmptyRelation

docs/changelog/136959.yaml

@@ -0,0 +1,6 @@
+pr: 136959
+summary: "ESQL: Support `StubRelation` with empty output in `PropagateEmptyRelation`"
+area: ES|QL
+type: bug
+issues:
+  - 136851

x-pack/plugin/esql/qa/testFixtures/src/main/resources/inlinestats.csv-spec

@@ -4166,3 +4166,39 @@ from employees
 c:long
 1
 ;
+
+
+// https://github.com/elastic/elasticsearch/issues/136851
+propagateEmptyStubRelation
+required_capability: fix_propagate_empty_stub_relation
+required_capability: fork_v9
+
+from countries_bbox_web 
+| limit 307 
+| drop `id` 
+| drop name 
+| inline stats  `shape` = count(*), UKmoYKgHt = count(*), jEHxaMcLul = count(*), VRjIowfF = count(*) 
+| drop UKmoYKgHt, UKmoYKgH*, shape 
+| FORK  (where  NOT true) (where  NOT false) (where  NOT VRjIowfF >= 50) (where jEHxaMcLul > 50 AND true | where  NOT false AND jEHxaMcLul == 50  | where false OR false OR  NOT false AND  NOT false) (where false AND  NOT false) 
+| WHERE _fork == "fork3" 
+| DROP _fork 
+| stats  oYhFLywK = avg(VRjIowfF) + count(VRjIowfF) + max(VRjIowfF), sygFReMcRSnf = count(*), WmxejCVu = count(*), REnkUKYklmVO = absent(VRjIowfF + VRjIowfF)
+;
+
+oYhFLywK:double | sygFReMcRSnf:long | WmxejCVu:long | REnkUKYklmVO:boolean
+248.0           | 248               | 248           | false
+;
+
+
+propagateEmptyStubRelation2
+required_capability: fix_propagate_empty_stub_relation
+
+ROW a = 12 
+| drop a 
+| inline stats b = count(*)
+;
+
+b:long
+0
+;
+

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java

@@ -1549,6 +1549,11 @@ public enum Cap {
          */
         LOOKUP_JOIN_SEMANTIC_FILTER_DEDUP,
 
+        /**
+         * Support for PropagateEmptyRelation with StubRelation when output is empty
+         */
+        FIX_PROPAGATE_EMPTY_STUB_RELATION(INLINESTATS_V11.enabled),
+
         // Last capability should still have a comma for fewer merge conflicts when adding new ones :)
         // This comment prevents the semicolon from being on the previous capability when Spotless formats the file.
         ;

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/PropagateEmptyRelation.java

@@ -22,6 +22,7 @@
 import org.elasticsearch.xpack.esql.plan.logical.Aggregate;
 import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
 import org.elasticsearch.xpack.esql.plan.logical.UnaryPlan;
+import org.elasticsearch.xpack.esql.plan.logical.join.StubRelation;
 import org.elasticsearch.xpack.esql.plan.logical.local.LocalRelation;
 import org.elasticsearch.xpack.esql.plan.logical.local.LocalSupplier;
 import org.elasticsearch.xpack.esql.planner.PlannerUtils;
@@ -41,7 +42,8 @@ public PropagateEmptyRelation() {
     @Override
     protected LogicalPlan rule(UnaryPlan plan, LogicalOptimizerContext ctx) {
         LogicalPlan p = plan;
-        if (plan.child() instanceof LocalRelation local && local.hasEmptySupplier()) {
+        if (plan.child() instanceof LocalRelation local && local.hasEmptySupplier()
+            || plan.child() instanceof StubRelation stub && stub.output().isEmpty()) {
             // only care about non-grouped aggs might return something (count)
             if (plan instanceof Aggregate agg && agg.groupings().isEmpty()) {
                 List<Block> emptyBlocks = aggsFromEmpty(ctx.foldCtx(), agg.aggregates());

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizerTests.java

@@ -9215,6 +9215,47 @@ public void testLookupJoinExpressionSwapped() {
         as(join.right(), EsRelation.class);
     }
 
+    /**
+     * <pre>{@code
+     * EsqlProject[[b{r}#6]]
+     * \_Eval[[0[LONG] AS b#6]]
+     *   \_Limit[1000[INTEGER],false,false]
+     *     \_LocalRelation[[a{r}#3],Page{blocks=[IntVectorBlock[vector=ConstantIntVector[positions=1, value=12]]]}]
+     * }</pre>
+     */
+    public void testPropagateEmptyStubRelation() {
+        var query = """
+            ROW a = 12
+            | drop a
+            | inline stats b = count(*)
+            """;
+        if (releaseBuildForInlineStats(query)) {
+            return;
+        }
+        var plan = optimizedPlan(query);
+
+        var project = as(plan, EsqlProject.class);
+        assertThat(Expressions.names(project.projections()), contains("b"));
+
+        var eval = as(project.child(), Eval.class);
+        assertThat(Expressions.names(eval.fields()), contains("b"));
+        var alias = as(eval.fields().getFirst(), Alias.class);
+        assertThat(alias.name(), equalTo("b"));
+        // The count(*) on empty stub should be 0
+        var literal = as(alias.child(), Literal.class);
+        assertThat(literal.value(), equalTo(0L));
+
+        var limit = asLimit(eval.child(), 1000, false);
+
+        var localRelation = as(limit.child(), LocalRelation.class);
+        assertThat(Expressions.names(localRelation.output()), contains("a"));
+
+        // Verify the local relation has one row with a = 12
+        Page page = localRelation.supplier().get();
+        assertThat(page.getBlockCount(), is(1));
+        assertThat(page.getPositionCount(), is(1));
+    }
+
     /**
      *
      * Project[[_meta_field{f}#15, emp_no{f}#9, first_name{f}#10, gender{f}#11, hire_date{f}#16, job{f}#17, job.raw{f}#18, la