ES|QL: Fix wrong pruning of plans with no output columns

docs/changelog/133405.yaml

@@ -0,0 +1,5 @@
+pr: 133405
+summary: Fix wrong pruning of plans with no output columns
+area: ES|QL
+type: bug
+issues: []

server/src/main/resources/transport/definitions/referable/esql_plan_with_no_columns.csv

@@ -0,0 +1 @@
+9186000

server/src/main/resources/transport/upper_bounds/9.3.csv

@@ -1 +1 @@
-initial_9.2.0,9185000
+esql_plan_with_no_columns,9186000

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/data/BasicPageTests.java

@@ -51,6 +51,15 @@ public void testEqualityAndHashCodeSmallInput() {
         );
         in.releaseBlocks();
 
+        in = new Page(10);
+        EqualsHashCodeTestUtils.checkEqualsAndHashCode(
+            in,
+            page -> new Page(10),
+            page -> new Page(8, blockFactory.newConstantIntBlockWith(1, 8)),
+            Page::releaseBlocks
+        );
+        in.releaseBlocks();
+
         in = new Page(blockFactory.newIntArrayVector(new int[] {}, 0).asBlock());
         EqualsHashCodeTestUtils.checkEqualsAndHashCode(
             in,
@@ -133,8 +142,8 @@ public void testEqualityAndHashCode() throws IOException {
             return new Page(blocks);
         };
 
-        int positions = randomIntBetween(1, 512);
-        int blockCount = randomIntBetween(1, 256);
+        int positions = randomIntBetween(0, 512);
+        int blockCount = randomIntBetween(0, 256);
         Block[] blocks = new Block[blockCount];
         for (int blockIndex = 0; blockIndex < blockCount; blockIndex++) {
             blocks[blockIndex] = switch (randomInt(9)) {
@@ -198,6 +207,16 @@ public void testAppend() {
         page2.releaseBlocks();
     }
 
+    public void testAppendToEmpty() {
+        Page page1 = new Page(10);
+        Page page2 = page1.appendBlock(blockFactory.newLongArrayVector(LongStream.range(0, 10).toArray(), 10).asBlock());
+        assertThat(0, is(page1.getBlockCount()));
+        assertThat(1, is(page2.getBlockCount()));
+        LongBlock block1 = page2.getBlock(0);
+        IntStream.range(0, 10).forEach(i -> assertThat((long) i, is(block1.getLong(i))));
+        page2.releaseBlocks();
+    }
+
     public void testPageSerializationSimple() throws IOException {
         IntVector toFilter = blockFactory.newIntArrayVector(IntStream.range(0, 20).toArray(), 20);
         Page origPage = new Page(
@@ -248,6 +267,22 @@ public void testPageSerializationSimple() throws IOException {
         }
     }
 
+    public void testPageSerializationEmpty() throws IOException {
+        Page origPage = new Page(10);
+        try {
+            Page deserPage = serializeDeserializePage(origPage);
+            try {
+                EqualsHashCodeTestUtils.checkEqualsAndHashCode(origPage, unused -> deserPage);
+                assertEquals(origPage.getBlockCount(), deserPage.getBlockCount());
+                assertEquals(origPage.getPositionCount(), deserPage.getPositionCount());
+            } finally {
+                deserPage.releaseBlocks();
+            }
+        } finally {
+            origPage.releaseBlocks();
+        }
+    }
+
     public void testSerializationListPages() throws IOException {
         final int positions = randomIntBetween(1, 64);
         List<Page> origPages = List.of(
@@ -265,7 +300,8 @@ public void testSerializationListPages() throws IOException {
                     positions
                 )
             ),
-            new Page(blockFactory.newConstantBytesRefBlockWith(new BytesRef("Hello World"), positions))
+            new Page(blockFactory.newConstantBytesRefBlockWith(new BytesRef("Hello World"), positions)),
+            new Page(10)
         );
         try {
             EqualsHashCodeTestUtils.checkEqualsAndHashCode(origPages, page -> {

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/EsqlTestUtils.java

@@ -36,6 +36,7 @@
 import org.elasticsearch.compute.data.DoubleBlock;
 import org.elasticsearch.compute.data.IntBlock;
 import org.elasticsearch.compute.data.LongBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.lucene.DataPartitioning;
 import org.elasticsearch.core.PathUtils;
 import org.elasticsearch.core.SuppressForbidden;
@@ -498,7 +499,11 @@ public static LogicalPlan emptySource() {
     }
 
     public static LogicalPlan localSource(BlockFactory blockFactory, List<Attribute> fields, List<Object> row) {
-        return new LocalRelation(Source.EMPTY, fields, LocalSupplier.of(BlockUtils.fromListRow(blockFactory, row)));
+        return new LocalRelation(
+            Source.EMPTY,
+            fields,
+            LocalSupplier.of(row.isEmpty() ? new Page(0) : new Page(BlockUtils.fromListRow(blockFactory, row)))
+        );
     }
 
     public static <T> T as(Object node, Class<T> type) {

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/generator/EsqlQueryGenerator.java

@@ -11,6 +11,7 @@
 import org.elasticsearch.xpack.esql.generator.command.CommandGenerator;
 import org.elasticsearch.xpack.esql.generator.command.pipe.ChangePointGenerator;
 import org.elasticsearch.xpack.esql.generator.command.pipe.DissectGenerator;
+import org.elasticsearch.xpack.esql.generator.command.pipe.DropAllGenerator;
 import org.elasticsearch.xpack.esql.generator.command.pipe.DropGenerator;
 import org.elasticsearch.xpack.esql.generator.command.pipe.EnrichGenerator;
 import org.elasticsearch.xpack.esql.generator.command.pipe.EvalGenerator;
@@ -63,6 +64,7 @@ public class EsqlQueryGenerator {
         ChangePointGenerator.INSTANCE,
         DissectGenerator.INSTANCE,
         DropGenerator.INSTANCE,
+        DropAllGenerator.INSTANCE,
         EnrichGenerator.INSTANCE,
         EvalGenerator.INSTANCE,
         ForkGenerator.INSTANCE,

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/generator/command/pipe/DropAllGenerator.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.generator.command.pipe;
+
+import org.elasticsearch.xpack.esql.generator.Column;
+import org.elasticsearch.xpack.esql.generator.EsqlQueryGenerator;
+import org.elasticsearch.xpack.esql.generator.QueryExecutor;
+import org.elasticsearch.xpack.esql.generator.command.CommandGenerator;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class DropAllGenerator implements CommandGenerator {
+
+    public static final String DROP_ALL = "drop_all";
+    public static final String DROPPED_COLUMNS = "dropped_columns";
+
+    public static final CommandGenerator INSTANCE = new DropAllGenerator();
+
+    @Override
+    public CommandDescription generate(
+        List<CommandDescription> previousCommands,
+        List<Column> previousOutput,
+        QuerySchema schema,
+        QueryExecutor executor
+    ) {
+        Set<String> droppedColumns = new HashSet<>();
+        String name = EsqlQueryGenerator.randomStringField(previousOutput);
+        if (name == null || name.isEmpty()) {
+            return CommandGenerator.EMPTY_DESCRIPTION;
+        }
+
+        String cmdString = " | keep " + name + " | drop " + name;
+        return new CommandDescription(DROP_ALL, this, cmdString, Map.ofEntries(Map.entry(DROPPED_COLUMNS, droppedColumns)));
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public ValidationResult validateOutput(
+        List<CommandDescription> previousCommands,
+        CommandDescription commandDescription,
+        List<Column> previousColumns,
+        List<List<Object>> previousOutput,
+        List<Column> columns,
+        List<List<Object>> output
+    ) {
+        if (commandDescription == EMPTY_DESCRIPTION) {
+            return VALIDATION_OK;
+        }
+
+        if (columns.size() > 0) {
+            return new ValidationResult(
+                false,
+                "Expecting no columns, got [" + columns.stream().map(Column::name).collect(Collectors.joining(", ")) + "]"
+            );
+        }
+
+        return VALIDATION_OK;
+    }
+
+}

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

@@ -49,29 +49,86 @@ b:integer | x:integer
 ;
 
 dropAllColumns
-from employees | keep height | drop height | eval x = 1;
+required_capability: fix_no_columns
+from employees | limit 4 | keep height | drop height | eval x = 1;
 
 x:integer
+1
+1
+1
+1
+;
+
+dropEvalKeep
+required_capability: fix_no_columns
+from employees | drop salary | eval salary = 1 | keep salary | limit 4;
+
+salary:integer
+1
+1
+1
+1
 ;
 
+
+dropEvalStats
+required_capability: fix_no_columns
+from mv_sample* | drop `client_ip`, message | eval  `event_duration` = "foo", @timestamp = 1 | stats max(@timestamp) by event_duration;
+
+max(@timestamp):integer | event_duration:keyword
+1                       | foo
+; 
+
+
+
+dropAllColumnsIndexPattern
+required_capability: fix_no_columns
+from emp* | drop languages | eval languages = 123 | keep languages | limit 4;
+
+languages:integer
+123
+123
+123
+123
+;
+
+
+dropAllColumnsWithMetadata
+required_capability: fix_no_columns
+from employees metadata _index | drop languages | eval languages = 123 | keep languages | limit 4;
+
+languages:integer
+123
+123
+123
+123
+;
+
+
 dropAllColumns_WithLimit
+required_capability: fix_no_columns
 from employees | keep height | drop height | eval x = 1 | limit 3;
 
 x:integer
+1
+1
+1
 ;
 
 dropAllColumns_WithCount
+required_capability: fix_no_columns
 from employees | keep height | drop height | eval x = 1 | stats c=count(x);
 
 c:long
-0
+100
 ;
 
 dropAllColumns_WithStats
+required_capability: fix_no_columns
 from employees | keep height | drop height | eval x = 1 | stats c=count(x), mi=min(x), s=sum(x);
 
-c:l|mi:i|s:l
-0  |null|null
+c:l  | mi:i | s:l
+100  | 1    | 100
 ;
 
 

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

@@ -5287,6 +5287,21 @@ null           | null             | corge          | null           | null
 null           | null             | fred           | null           | null
 ;
 
+lookupAfterDropAllColumns
+required_capability: fix_no_columns
+required_capability: join_lookup_v12
+FROM languages
+| DROP language_code
+| EVAL language_code = 3
+| LOOKUP JOIN languages_lookup ON language_code
+| LIMIT 3
+;
+
+ language_code:integer | language_name:keyword 
+3                      |Spanish
+3                      |Spanish
+3                      |Spanish
+;
 
 lookupJoinWithPushableFilterOnRight
 required_capability: join_lookup_v12

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

@@ -246,3 +246,13 @@ emp_no:integer
 10081
 // end::sampleForDocs-result[]
 ;
+
+
+sampleStatsEval
+required_capability: fix_no_columns
+required_capability: sample_v3
+FROM employees | SAMPLE 0.5 | LIMIT 10 | STATS count = COUNT() | EVAL is_expected = count > 0;
+
+count:long | is_expected:boolean
+10         | true
+;

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterLookupJoinIT.java

@@ -445,7 +445,7 @@ public void testLookupJoinEmptyIndex() throws IOException {
         setSkipUnavailable(REMOTE_CLUSTER_1, randomBoolean());
 
         Exception ex;
-        for (String index : List.of("values_lookup", "values_lookup_map", "values_lookup_map_lookup")) {
+        for (String index : List.of("values_lookup", "values_lookup_map_lookup")) {
             ex = expectThrows(
                 VerificationException.class,
                 () -> runQuery("FROM logs-* | LOOKUP JOIN " + index + " ON v | KEEP v", randomBoolean())
@@ -457,6 +457,18 @@ public void testLookupJoinEmptyIndex() throws IOException {
             );
             assertThat(ex.getMessage(), containsString("Unknown column [v] in right side of join"));
         }
+
+        ex = expectThrows(
+            VerificationException.class,
+            () -> runQuery("FROM logs-* | LOOKUP JOIN values_lookup_map ON v | KEEP v", randomBoolean())
+        );
+        assertThat(
+            ex.getMessage(),
+            containsString(
+                "Lookup Join requires a single lookup mode index; "
+                    + "[values_lookup_map] resolves to [values_lookup_map] in [standard] mode"
+            )
+        );
     }
 
     public void testLookupJoinIndexMode() throws IOException {

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/EsqlActionIT.java

@@ -959,15 +959,15 @@ public void testDropAllColumns() {
             logger.info(results);
             assertThat(results.columns(), hasSize(1));
             assertThat(results.columns(), contains(new ColumnInfoImpl("a", "integer", null)));
-            assertThat(getValuesList(results), is(empty()));
+            assertThat(getValuesList(results).size(), is(40));
         }
     }
 
     public void testDropAllColumnsWithStats() {
         try (EsqlQueryResponse results = run("from test | stats g = count(data) | drop g")) {
             logger.info(results);
             assertThat(results.columns(), is(empty()));
-            assertThat(getValuesList(results), is(empty()));
+            assertThat(getValuesList(results).size(), is(1));
         }
     }
 

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

@@ -1587,7 +1587,13 @@ public enum Cap {
         /**
          * Support for requesting the "_tsid" metadata field.
          */
-        METADATA_TSID_FIELD;
+        METADATA_TSID_FIELD,
+
+        /**
+         * Fix management of plans with no columns
+         * https://github.com/elastic/elasticsearch/issues/120272
+         */
+        FIX_NO_COLUMNS;
 
         private final boolean enabled;
 

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

@@ -12,6 +12,7 @@
 import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.compute.data.AggregateMetricDoubleBlockBuilder;
 import org.elasticsearch.compute.data.Block;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.core.Strings;
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.logging.Logger;
@@ -472,7 +473,7 @@ private LocalRelation tableMapAsRelation(Source source, Map<String, Column> mapT
                 // prepare the block for the supplier
                 blocks[i++] = column.values();
             }
-            LocalSupplier supplier = LocalSupplier.of(blocks);
+            LocalSupplier supplier = LocalSupplier.of(blocks.length > 0 ? new Page(blocks) : new Page(0));
             return new LocalRelation(source, attributes, supplier);
         }
     }