QL: EQL and ESQL to use only the necessary fields in the internal field_caps calls (#98987)

* Use only the necessary fields in field_caps calls from QL

docs/changelog/98987.yaml

@@ -0,0 +1,6 @@
+pr: 98987
+summary: EQL and ESQL to use only the necessary fields in the internal `field_caps`
+  calls
+area: EQL
+type: enhancement
+issues: []

x-pack/plugin/eql/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/eql/10_basic.yml

@@ -454,3 +454,27 @@ setup:
         body:
           query: 'sequence with maxspan=10d [network where user == "ADMIN"] ![network where user == "SYSTEM"] [network where user == "ADMIN"]'
   - match: {hits.total.value: 0}
+
+---
+"Error message missing column - no suggestion":
+
+  - do:
+      catch: "bad_request"
+      eql.search:
+        index: eql_test
+        body:
+          query: 'sequence with maxspan=10d [network where used == "ADMIN"] [network where id == 123]'
+  - match: { error.root_cause.0.type: "verification_exception" }
+  - match: { error.root_cause.0.reason: "Found 1 problem\nline 1:42: Unknown column [used]" }
+
+---
+"Error message missing column - did you mean functionality":
+
+  - do:
+      catch: "bad_request"
+      eql.search:
+        index: eql_test
+        body:
+          query: 'sequence with maxspan=10d [network where user == "ADMIN"] ![network where used == "SYSTEM"]'
+  - match: { error.root_cause.0.type: "verification_exception" }
+  - match: { error.root_cause.0.reason: "Found 1 problem\nline 1:75: Unknown column [used], did you mean [user]?" }

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

@@ -22,11 +22,16 @@ 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.UnresolvedAttribute;
 import org.elasticsearch.xpack.ql.expression.function.FunctionRegistry;
 import org.elasticsearch.xpack.ql.index.IndexResolver;
 import org.elasticsearch.xpack.ql.plan.logical.LogicalPlan;
 
+import java.util.LinkedHashSet;
+import java.util.Set;
+
 import static org.elasticsearch.xpack.ql.util.ActionListeners.map;
+import static org.elasticsearch.xpack.ql.util.StringUtils.WILDCARD;
 
 public class EqlSession {
 
@@ -116,14 +121,27 @@ public class EqlSession {
             listener.onFailure(new TaskCancelledException("cancelled"));
             return;
         }
+        Set<String> fieldNames = fieldNames(parsed);
         indexResolver.resolveAsMergedMapping(
             indexWildcard,
+            fieldNames,
             configuration.indicesOptions(),
             configuration.runtimeMappings(),
             map(listener, r -> preAnalyzer.preAnalyze(parsed, r))
         );
     }
 
+    static Set<String> fieldNames(LogicalPlan parsed) {
+        Set<String> fieldNames = new LinkedHashSet<>();
+        parsed.forEachExpressionDown(UnresolvedAttribute.class, ua -> {
+            fieldNames.add(ua.name());
+            if (ua.name().endsWith(WILDCARD) == false) {
+                fieldNames.add(ua.name() + ".*");
+            }
+        });
+        return fieldNames.isEmpty() ? IndexResolver.ALL_FIELDS : fieldNames;
+    }
+
     private LogicalPlan postAnalyze(LogicalPlan verified) {
         return postAnalyzer.postAnalyze(verified, configuration);
     }

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

@@ -0,0 +1,492 @@
+/*
+ * 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.eql.session;
+
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.eql.parser.EqlParser;
+
+import java.util.Set;
+
+import static org.hamcrest.Matchers.equalTo;
+
+public class IndexResolverFieldNamesTests extends ESTestCase {
+
+    private static final EqlParser parser = new EqlParser();
+
+    public void testSimpleQueryEqual() {
+        assertFieldNames(
+            """
+                process where serial_event_id == 1""",
+            Set.of("serial_event_id.*", "serial_event_id", "event.category.*", "event.category", "@timestamp.*", "@timestamp")
+        );
+    }
+
+    public void testSimpleQueryHeadSix() {
+        assertFieldNames("""
+            process where true | head 6""", Set.of("event.category.*", "event.category", "@timestamp.*", "@timestamp"));
+    }
+
+    public void testProcessWhereFalse() {
+        assertFieldNames("""
+            process where false""", Set.of("event.category.*", "event.category", "@timestamp.*", "@timestamp"));
+    }
+
+    public void testProcessNameInexistent() {
+        assertFieldNames(
+            """
+                process where process_name : "impossible name" or (serial_event_id < 4.5 and serial_event_id >= 3.1)""",
+            Set.of(
+                "process_name.*",
+                "process_name",
+                "serial_event_id.*",
+                "serial_event_id",
+                "event.category.*",
+                "event.category",
+                "@timestamp.*",
+                "@timestamp"
+            )
+        );
+    }
+
+    public void testSerialEventIdLteAndGt() {
+        assertFieldNames(
+            """
+                process where serial_event_id<=8 and serial_event_id > 7""",
+            Set.of("serial_event_id.*", "serial_event_id", "event.category.*", "event.category", "@timestamp.*", "@timestamp")
+        );
+    }
+
+    public void testMinusOneLtExitCode() {
+        assertFieldNames(
+            """
+                process where -1 < exit_code""",
+            Set.of("exit_code.*", "exit_code", "event.category.*", "event.category", "@timestamp.*", "@timestamp")
+        );
+    }
+
+    public void testNotExitCodeGtWithHead1() {
+        assertFieldNames(
+            """
+                process where not (exit_code > -1)
+                  and serial_event_id in (58, 64, 69, 74, 80, 85, 90, 93, 94)
+                | head 10""",
+            Set.of(
+                "exit_code.*",
+                "exit_code",
+                "serial_event_id.*",
+                "serial_event_id",
+                "event.category.*",
+                "event.category",
+                "@timestamp.*",
+                "@timestamp"
+            )
+        );
+    }
+
+    public void testProcessWithMultipleConditions1() {
+        assertFieldNames(
+            """
+                process where (serial_event_id<=8 and serial_event_id > 7) and (opcode==3 and opcode>2)""",
+            Set.of(
+                "opcode.*",
+                "opcode",
+                "serial_event_id.*",
+                "serial_event_id",
+                "event.category.*",
+                "event.category",
+                "@timestamp.*",
+                "@timestamp"
+            )
+        );
+    }
+
+    public void testWildcardAndMultipleConditions1() {
+        assertFieldNames(
+            """
+                file where file_path:"x"
+                  and opcode in (0,1,2) and user_name:\"vagrant\"""",
+            Set.of(
+                "user_name.*",
+                "user_name",
+                "opcode.*",
+                "opcode",
+                "file_path.*",
+                "file_path",
+                "event.category.*",
+                "event.category",
+                "@timestamp.*",
+                "@timestamp"
+            )
+        );
+    }
+
+    public void testSequenceOneOneMatch() {
+        assertFieldNames(
+            """
+                sequence
+                  [process where serial_event_id == 1]
+                  [process where serial_event_id == 2]""",
+            Set.of("serial_event_id.*", "serial_event_id", "event.category.*", "event.category", "@timestamp.*", "@timestamp")
+        );
+    }
+
+    public void testSequenceOneManyMany_Runs() {
+        assertFieldNames(
+            """
+                sequence
+                  [process where serial_event_id == 1]
+                  [process where true] with runs=2""",
+            Set.of("serial_event_id.*", "serial_event_id", "event.category.*", "event.category", "@timestamp.*", "@timestamp")
+        );
+    }
+
+    public void testTwoSequencesWithKeys() {
+        assertFieldNames(
+            """
+                sequence
+                  [process where true]        by unique_pid
+                  [process where opcode == 1] by unique_ppid""",
+            Set.of(
+                "opcode.*",
+                "opcode",
+                "unique_ppid.*",
+                "unique_ppid",
+                "unique_pid.*",
+                "unique_pid",
+                "event.category.*",
+                "event.category",
+                "@timestamp.*",
+                "@timestamp"
+            )
+        );
+    }
+
+    public void testTwoSequencesWithTwoKeys() {
+        assertFieldNames(
+            """
+                sequence
+                  [process where true]        by unique_pid,  process_path
+                  [process where opcode == 1] by unique_ppid, parent_process_path""",
+            Set.of(
+                "opcode.*",
+                "opcode",
+                "unique_ppid.*",
+                "unique_ppid",
+                "unique_pid.*",
+                "unique_pid",
+                "process_path.*",
+                "process_path",
+                "parent_process_path.*",
+                "parent_process_path",
+                "event.category.*",
+                "event.category",
+                "@timestamp.*",
+                "@timestamp"
+            )
+        );
+    }
+
+    public void testFourSequencesByPidWithUntil1() {
+        assertFieldNames(
+            """
+                sequence
+                  [process where opcode == 1] by unique_pid
+                  [file where opcode == 0]    by unique_pid
+                  [file where opcode == 0]    by unique_pid
+                  [file where opcode == 0]    by unique_pid
+                until
+                  [file where opcode == 2]    by unique_pid""",
+            Set.of("opcode.*", "opcode", "unique_pid.*", "unique_pid", "event.category.*", "event.category", "@timestamp.*", "@timestamp")
+        );
+    }
+
+    public void testSequencesOnDifferentEventTypesWithBy() {
+        assertFieldNames(
+            """
+                sequence
+                  [file where opcode==0 and file_name:"svchost.exe"] by unique_pid
+                  [process where opcode == 1] by unique_ppid""",
+            Set.of(
+                "opcode.*",
+                "opcode",
+                "unique_ppid.*",
+                "unique_ppid",
+                "unique_pid.*",
+                "unique_pid",
+                "file_name.*",
+                "file_name",
+                "event.category.*",
+                "event.category",
+                "@timestamp.*",
+                "@timestamp"
+            )
+        );
+    }
+
+    public void testMultipleConditions2() {
+        assertFieldNames(
+            """
+                process where opcode == 1
+                  and process_name in ("net.exe", "net1.exe")
+                  and not (parent_process_name : "net.exe"
+                  and process_name : "net1.exe")
+                  and command_line : "*group *admin*" and command_line != \"*x*\"""",
+            Set.of(
+                "opcode.*",
+                "opcode",
+                "process_name.*",
+                "process_name",
+                "parent_process_name.*",
+                "parent_process_name",
+                "command_line.*",
+                "command_line",
+                "event.category.*",
+                "event.category",
+                "@timestamp.*",
+                "@timestamp"
+            )
+        );
+    }
+
+    public void testTwoSequencesWithKeys2() {
+        assertFieldNames(
+            """
+                sequence
+                  [file where file_name:"lsass.exe"] by file_path,process_path
+                  [process where true] by process_path,parent_process_path""",
+            Set.of(
+                "file_name.*",
+                "file_name",
+                "file_path.*",
+                "file_path",
+                "process_path.*",
+                "process_path",
+                "parent_process_path.*",
+                "parent_process_path",
+                "event.category.*",
+                "event.category",
+                "@timestamp.*",
+                "@timestamp"
+            )
+        );
+    }
+
+    public void testEndsWithAndCondition() {
+        assertFieldNames(
+            """
+                file where opcode==0 and serial_event_id == 88 and startsWith~("explorer.exeaAAAA", "EXPLORER.exe")""",
+            Set.of(
+                "opcode.*",
+                "opcode",
+                "serial_event_id.*",
+                "serial_event_id",
+                "event.category.*",
+                "event.category",
+                "@timestamp.*",
+                "@timestamp"
+            )
+        );
+    }
+
+    public void testStringContains2() {
+        assertFieldNames(
+            """
+                file where opcode==0 and stringContains("ABCDEFGHIexplorer.exeJKLMNOP", file_name)""",
+            Set.of("opcode.*", "opcode", "file_name.*", "file_name", "event.category.*", "event.category", "@timestamp.*", "@timestamp")
+        );
+    }
+
+    public void testConcatCaseInsensitive() {
+        assertFieldNames(
+            "process where concat(serial_event_id, \":\", process_name, opcode) : \"x\"",
+            Set.of(
+                "opcode.*",
+                "opcode",
+                "process_name.*",
+                "process_name",
+                "serial_event_id.*",
+                "serial_event_id",
+                "event.category.*",
+                "event.category",
+                "@timestamp.*",
+                "@timestamp"
+            )
+        );
+    }
+
+    public void testCidrMatch4() {
+        assertFieldNames(
+            """
+                network where cidrMatch(source_address, "0.0.0.0/0")""",
+            Set.of("source_address.*", "source_address", "event.category.*", "event.category", "@timestamp.*", "@timestamp")
+        );
+    }
+
+    public void testNumberStringConversion5() {
+        assertFieldNames(
+            """
+                any where number(string(serial_event_id), 16) == 17""",
+            Set.of("serial_event_id.*", "serial_event_id", "@timestamp.*", "@timestamp")
+        );
+    }
+
+    public void testSimpleRegex() {
+        assertFieldNames(
+            "process where command_line regex \".*\"",
+            Set.of("command_line.*", "command_line", "event.category.*", "event.category", "@timestamp.*", "@timestamp")
+        );
+    }
+
+    public void testSequenceWithOptionalUserDomain() {
+        assertFieldNames(
+            """
+                sequence by ?user_domain [process where true] [registry where true]""",
+            Set.of("user_domain.*", "user_domain", "event.category.*", "event.category", "@timestamp.*", "@timestamp")
+        );
+    }
+
+    public void testTwoSequencesWithTwoKeys_AndOptionals() {
+        assertFieldNames(
+            """
+                sequence by ?x
+                  [process where true]        by unique_pid,  process_path,        ?z
+                  [process where opcode == 1] by unique_ppid, parent_process_path, ?w""",
+            Set.of(
+                "opcode.*",
+                "opcode",
+                "x.*",
+                "x",
+                "parent_process_path.*",
+                "parent_process_path",
+                "process_path.*",
+                "process_path",
+                "unique_pid.*",
+                "unique_pid",
+                "unique_ppid.*",
+                "unique_ppid",
+                "z.*",
+                "z",
+                "w.*",
+                "w",
+                "event.category.*",
+                "event.category",
+                "@timestamp.*",
+                "@timestamp"
+            )
+        );
+    }
+
+    public void testOptionalDefaultNullValueFieldEqualNull() {
+        assertFieldNames(
+            """
+                OPTIONAL where ?optional_field_default_null == null""",
+            Set.of(
+                "optional_field_default_null.*",
+                "optional_field_default_null",
+                "event.category.*",
+                "event.category",
+                "@timestamp.*",
+                "@timestamp"
+            )
+        );
+    }
+
+    public void testSequenceOptionalFieldAsQueryKeys() {
+        assertFieldNames(
+            """
+                sequence by ?x, transID
+                  [ERROR where true] by ?x
+                  [OPTIONAL where true] by ?y""",
+            Set.of("x.*", "x", "y.*", "y", "transID.*", "transID", "event.category.*", "event.category", "@timestamp.*", "@timestamp")
+        );
+    }
+
+    public void testSequenceAllKeysOptional() {
+        assertFieldNames(
+            """
+                sequence by ?process.entity_id, ?process.pid
+                  [process where transID == 2]
+                  [file where transID == 0] with runs=2""",
+            Set.of(
+                "process.entity_id.*",
+                "process.entity_id",
+                "process.pid.*",
+                "process.pid",
+                "transID.*",
+                "transID",
+                "event.category.*",
+                "event.category",
+                "@timestamp.*",
+                "@timestamp"
+            )
+        );
+    }
+
+    public void testMultipleMissing1() {
+        assertFieldNames("""
+            sequence with maxspan=1s
+                [ test4 where tag == "A" ]
+                [ test4 where tag == "B" ]
+               ![ test4 where tag == "M1"]
+                [ test4 where tag == "C" ]
+               ![ test4 where tag == "M2"]
+                [ test4 where tag == "D" ]""", Set.of("tag.*", "tag", "event.category.*", "event.category", "@timestamp.*", "@timestamp"));
+    }
+
+    public void testWithByKey_runs() {
+        assertFieldNames(
+            """
+                sequence by k1 with maxspan=1s
+                    [ test5 where tag == "normal" ] by k2 with runs=2
+                   ![ test5 where tag == "missing" ] by k2
+                    [ test5 where tag == "normal" ] by k2""",
+            Set.of("tag.*", "tag", "k1.*", "k1", "k2.*", "k2", "event.category.*", "event.category", "@timestamp.*", "@timestamp")
+        );
+    }
+
+    public void testComplexFiltersWithSample() {
+        assertFieldNames(
+            """
+                sample by host
+                    [any where uptime > 0 and host == "doom" and (uptime > 15 or bool == true)] by os
+                    [any where port > 100 and ip == "10.0.0.5" or op_sys : "REDHAT"] by op_sys
+                    [any where bool == true] by os""",
+            Set.of(
+                "host.*",
+                "host",
+                "uptime.*",
+                "uptime",
+                "bool.*",
+                "bool",
+                "os.*",
+                "os",
+                "port.*",
+                "port",
+                "ip.*",
+                "ip",
+                "op_sys.*",
+                "op_sys"
+            )
+        );
+    }
+
+    public void testOptionalFieldAsKeyAndMultipleConditions() {
+        assertFieldNames(
+            """
+                sample by ?x, ?y
+                    [failure where (?x == null or ?y == null) and id == 17]
+                    [success where (?y == null and ?x == null) and id == 18]""",
+            Set.of("x.*", "x", "y.*", "y", "id.*", "id", "event.category.*", "event.category")
+        );
+    }
+
+    private void assertFieldNames(String query, Set<String> expected) {
+        Set<String> fieldNames = EqlSession.fieldNames(parser.createStatement(query));
+        assertThat(fieldNames, equalTo(expected));
+    }
+}

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/EnrichPolicyResolver.java

@@ -39,6 +39,7 @@ public class EnrichPolicyResolver {
         try (ThreadContext.StoredContext ignored = threadContext.stashWithOrigin(ClientHelper.ENRICH_ORIGIN)) {
             indexResolver.resolveAsMergedMapping(
                 EnrichPolicy.getBaseName(policyName),
+                IndexResolver.ALL_FIELDS,
                 false,
                 Map.of(),
                 listener.map(indexResult -> new EnrichPolicyResolution(policyName, policy, indexResult))

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlSession.java

@@ -11,6 +11,7 @@ import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.RefCountingListener;
+import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.xpack.esql.action.EsqlQueryRequest;
@@ -25,17 +26,27 @@ import org.elasticsearch.xpack.esql.optimizer.PhysicalOptimizerContext;
 import org.elasticsearch.xpack.esql.optimizer.PhysicalPlanOptimizer;
 import org.elasticsearch.xpack.esql.parser.EsqlParser;
 import org.elasticsearch.xpack.esql.parser.TypedParamValue;
+import org.elasticsearch.xpack.esql.plan.logical.Keep;
+import org.elasticsearch.xpack.esql.plan.logical.RegexExtract;
 import org.elasticsearch.xpack.esql.plan.physical.EstimatesRowSize;
 import org.elasticsearch.xpack.esql.plan.physical.FragmentExec;
 import org.elasticsearch.xpack.esql.plan.physical.PhysicalPlan;
 import org.elasticsearch.xpack.esql.planner.Mapper;
 import org.elasticsearch.xpack.ql.analyzer.TableInfo;
+import org.elasticsearch.xpack.ql.expression.Alias;
+import org.elasticsearch.xpack.ql.expression.Attribute;
+import org.elasticsearch.xpack.ql.expression.AttributeSet;
+import org.elasticsearch.xpack.ql.expression.MetadataAttribute;
+import org.elasticsearch.xpack.ql.expression.UnresolvedStar;
 import org.elasticsearch.xpack.ql.expression.function.FunctionRegistry;
 import org.elasticsearch.xpack.ql.index.IndexResolution;
 import org.elasticsearch.xpack.ql.index.IndexResolver;
 import org.elasticsearch.xpack.ql.index.MappingException;
 import org.elasticsearch.xpack.ql.plan.TableIdentifier;
+import org.elasticsearch.xpack.ql.plan.logical.Aggregate;
 import org.elasticsearch.xpack.ql.plan.logical.LogicalPlan;
+import org.elasticsearch.xpack.ql.plan.logical.Project;
+import org.elasticsearch.xpack.ql.util.Holder;
 
 import java.util.HashSet;
 import java.util.List;
@@ -43,9 +54,11 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiFunction;
+import java.util.stream.Collectors;
 
 import static org.elasticsearch.index.query.QueryBuilders.boolQuery;
 import static org.elasticsearch.xpack.ql.util.ActionListeners.map;
+import static org.elasticsearch.xpack.ql.util.StringUtils.WILDCARD;
 
 public class EsqlSession {
 
@@ -159,7 +172,8 @@ public class EsqlSession {
         } else if (preAnalysis.indices.size() == 1) {
             TableInfo tableInfo = preAnalysis.indices.get(0);
             TableIdentifier table = tableInfo.id();
-            indexResolver.resolveAsMergedMapping(table.index(), false, Map.of(), listener);
+            var fieldNames = fieldNames(parsed);
+            indexResolver.resolveAsMergedMapping(table.index(), fieldNames, false, Map.of(), listener);
         } else {
             try {
                 // occurs when dealing with local relations (row a = 1)
@@ -170,6 +184,80 @@ public class EsqlSession {
         }
     }
 
+    static Set<String> fieldNames(LogicalPlan parsed) {
+        if (false == parsed.anyMatch(plan -> plan instanceof Aggregate || plan instanceof Project)) {
+            // no explicit columns selection, for example "from employees"
+            return IndexResolver.ALL_FIELDS;
+        }
+
+        Holder<Boolean> projectAll = new Holder<>(false);
+        parsed.forEachExpressionDown(UnresolvedStar.class, us -> {// explicit "*" fields selection
+            if (projectAll.get()) {
+                return;
+            }
+            projectAll.set(true);
+        });
+        if (projectAll.get()) {
+            return IndexResolver.ALL_FIELDS;
+        }
+
+        AttributeSet references = new AttributeSet();
+        // "keep" attributes are special whenever a wildcard is used in their name
+        // ie "from test | eval lang = languages + 1 | keep *l" should consider both "languages" and "*l" as valid fields to ask for
+        AttributeSet keepCommandReferences = new AttributeSet();
+
+        parsed.forEachDown(p -> {// go over each plan top-down
+            if (p instanceof RegexExtract re) { // for Grok and Dissect
+                AttributeSet dissectRefs = p.references();
+                // don't add to the list of fields the extracted ones (they are not real fields in mappings)
+                dissectRefs.removeAll(re.extractedFields());
+                references.addAll(dissectRefs);
+                // also remove other down-the-tree references to the extracted fields
+                for (Attribute extracted : re.extractedFields()) {
+                    references.removeIf(attr -> matchByName(attr, extracted.qualifiedName(), false));
+                }
+            } else {
+                references.addAll(p.references());
+                if (p instanceof Keep) {
+                    keepCommandReferences.addAll(p.references());
+                }
+            }
+
+            // remove any already discovered UnresolvedAttributes that are in fact aliases defined later down in the tree
+            // for example "from test | eval x = salary | stats max = max(x) by gender"
+            // remove the UnresolvedAttribute "x", since that is an Alias defined in "eval"
+            p.forEachExpressionDown(Alias.class, alias -> {
+                // do not remove the UnresolvedAttribute that has the same name as its alias, ie "rename id = id"
+                // or the UnresolvedAttributes that are used in Functions that have aliases "STATS id = MAX(id)"
+                if (p.references().names().contains(alias.qualifiedName())) {
+                    return;
+                }
+                references.removeIf(attr -> matchByName(attr, alias.qualifiedName(), keepCommandReferences.contains(attr)));
+            });
+        });
+
+        // remove valid metadata attributes because they will be filtered out by the IndexResolver anyway
+        // otherwise, in some edge cases, we will fail to ask for "*" (all fields) instead
+        references.removeIf(a -> a instanceof MetadataAttribute || MetadataAttribute.isSupported(a.qualifiedName()));
+        Set<String> fieldNames = references.names();
+        if (fieldNames.isEmpty()) {
+            return IndexResolver.ALL_FIELDS;
+        } else {
+            fieldNames.addAll(
+                fieldNames.stream().filter(name -> name.endsWith(WILDCARD) == false).map(name -> name + ".*").collect(Collectors.toSet())
+            );
+            return fieldNames;
+        }
+    }
+
+    private static boolean matchByName(Attribute attr, String other, boolean skipIfPattern) {
+        boolean isPattern = Regex.isSimpleMatchPattern(attr.qualifiedName());
+        if (skipIfPattern && isPattern) {
+            return false;
+        }
+        return isPattern ? Regex.simpleMatch(attr.qualifiedName(), other) : attr.qualifiedName().equals(other);
+    }
+
     public void optimizedPlan(LogicalPlan logicalPlan, ActionListener<LogicalPlan> listener) {
         analyzedPlan(logicalPlan, map(listener, p -> {
             var plan = logicalPlanOptimizer.optimize(p);

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/session/IndexResolverFieldNamesTests.java

@@ -0,0 +1,1115 @@
+/*
+ * 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.session;
+
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.esql.parser.EsqlParser;
+
+import java.util.Set;
+
+import static org.elasticsearch.xpack.ql.index.IndexResolver.ALL_FIELDS;
+import static org.hamcrest.Matchers.equalTo;
+
+public class IndexResolverFieldNamesTests extends ESTestCase {
+
+    private static final EsqlParser parser = new EsqlParser();
+
+    public void testBasicFromCommand() {
+        assertFieldNames("from test", ALL_FIELDS);
+    }
+
+    public void testBasicFromCommandWithMetadata() {
+        assertFieldNames("from test [metadata _index, _id, _version]", ALL_FIELDS);
+    }
+
+    public void testSimple1() {
+        assertFieldNames(
+            "from employees | sort emp_no | keep emp_no, still_hired | limit 3",
+            Set.of("emp_no", "emp_no.*", "still_hired", "still_hired.*")
+        );
+    }
+
+    public void testDirectFilter() {
+        assertFieldNames(
+            "from employees | sort emp_no | where still_hired | keep emp_no | limit 3",
+            Set.of("emp_no", "emp_no.*", "still_hired", "still_hired.*")
+        );
+    }
+
+    public void testSort1() {
+        assertFieldNames(
+            "from employees | sort still_hired, emp_no | keep emp_no, still_hired | limit 3",
+            Set.of("emp_no", "emp_no.*", "still_hired", "still_hired.*")
+        );
+    }
+
+    public void testStatsBy() {
+        assertFieldNames(
+            "from employees | stats avg(salary) by still_hired | sort still_hired",
+            Set.of("salary", "salary.*", "still_hired", "still_hired.*")
+        );
+    }
+
+    public void testStatsByAlwaysTrue() {
+        assertFieldNames(
+            "from employees | where first_name is not null | eval always_true = starts_with(first_name, \"\") "
+                + "| stats avg(salary) by always_true",
+            Set.of("first_name", "first_name.*", "salary", "salary.*")
+        );
+    }
+
+    public void testStatsByAlwaysFalse() {
+        assertFieldNames(
+            "from employees | where first_name is not null "
+                + "| eval always_false = starts_with(first_name, \"nonestartwiththis\") "
+                + "| stats avg(salary) by always_false",
+            Set.of("first_name", "first_name.*", "salary", "salary.*")
+        );
+    }
+
+    public void testIn1() {
+        assertFieldNames(
+            "from employees | keep emp_no, is_rehired, still_hired "
+                + "| where is_rehired in (still_hired, true) | where is_rehired != still_hired",
+            Set.of("emp_no", "emp_no.*", "is_rehired", "is_rehired.*", "still_hired", "still_hired.*")
+        );
+    }
+
+    public void testConvertFromString1() {
+        assertFieldNames("""
+            from employees
+            | keep emp_no, is_rehired, first_name
+            | eval rehired_str = to_string(is_rehired)
+            | eval rehired_bool = to_boolean(rehired_str)
+            | eval all_false = to_boolean(first_name)
+            | drop first_name
+            | limit 5""", Set.of("emp_no", "emp_no.*", "is_rehired", "is_rehired.*", "first_name", "first_name.*"));
+    }
+
+    public void testConvertFromDouble1() {
+        assertFieldNames("""
+            from employees
+            | eval h_2 = height - 2.0, double2bool = to_boolean(h_2)
+            | where emp_no in (10036, 10037, 10038)
+            | keep emp_no, height, *2bool""", Set.of("height", "height.*", "emp_no", "emp_no.*", "h_2", "h_2.*", "*2bool.*", "*2bool"));
+        // TODO asking for more shouldn't hurt. Can we do better? ("h_2" shouldn't be in the list of fields)
+        // Set.of("height", "height.*", "emp_no", "emp_no.*", "*2bool.*", "*2bool"));
+    }
+
+    public void testConvertFromIntAndLong() {
+        assertFieldNames(
+            "from employees | keep emp_no, salary_change*"
+                + "| eval int2bool = to_boolean(salary_change.int), long2bool = to_boolean(salary_change.long) | limit 10",
+            Set.of(
+                "emp_no",
+                "emp_no.*",
+                "salary_change*",
+                "salary_change.int.*",
+                "salary_change.int",
+                "salary_change.long.*",
+                "salary_change.long"
+            )
+        );
+    }
+
+    public void testIntToInt() {
+        assertFieldNames("""
+            from employees
+            | where emp_no < 10002
+            | keep emp_no""", Set.of("emp_no", "emp_no.*"));
+    }
+
+    public void testLongToLong() {
+        assertFieldNames(
+            """
+                from employees
+                | where languages.long < avg_worked_seconds
+                | limit 1
+                | keep emp_no""",
+            Set.of("emp_no", "emp_no.*", "languages.long", "languages.long.*", "avg_worked_seconds", "avg_worked_seconds.*")
+        );
+    }
+
+    public void testDateToDate() {
+        assertFieldNames("""
+            from employees
+            | where birth_date < hire_date
+            | keep emp_no
+            | sort emp_no
+            | limit 1""", Set.of("birth_date", "birth_date.*", "emp_no", "emp_no.*", "hire_date", "hire_date.*"));
+    }
+
+    public void testTwoConditionsWithDefault() {
+        assertFieldNames("""
+            from employees
+            | eval type = case(languages <= 1, "monolingual", languages <= 2, "bilingual", "polyglot")
+            | keep emp_no, type
+            | limit 10""", Set.of("emp_no", "emp_no.*", "languages", "languages.*"));
+    }
+
+    public void testSingleCondition() {
+        assertFieldNames("""
+            from employees
+            | eval g = case(gender == "F", true)
+            | keep gender, g
+            | limit 10""", Set.of("gender", "gender.*"));
+    }
+
+    public void testConditionIsNull() {
+        assertFieldNames("""
+            from employees
+            | eval g = case(gender == "F", 1, languages > 1, 2, 3)
+            | keep gender, languages, g
+            | limit 25""", Set.of("gender", "gender.*", "languages", "languages.*"));
+    }
+
+    public void testEvalAssign() {
+        assertFieldNames(
+            "from employees | sort hire_date | eval x = hire_date | keep emp_no, x | limit 5",
+            Set.of("hire_date", "hire_date.*", "emp_no", "emp_no.*")
+        );
+    }
+
+    public void testMinMax() {
+        assertFieldNames("from employees | stats min = min(hire_date), max = max(hire_date)", Set.of("hire_date", "hire_date.*"));
+    }
+
+    public void testEvalDateTruncIntervalExpressionPeriod() {
+        assertFieldNames(
+            "from employees | sort hire_date | eval x = date_trunc(hire_date, 1 month) | keep emp_no, hire_date, x | limit 5",
+            Set.of("hire_date", "hire_date.*", "emp_no", "emp_no.*")
+        );
+    }
+
+    public void testEvalDateTruncGrouping() {
+        assertFieldNames("""
+            from employees
+            | eval y = date_trunc(hire_date, 1 year)
+            | stats count(emp_no) by y
+            | sort y
+            | keep y, count(emp_no)
+            | limit 5""", Set.of("hire_date", "hire_date.*", "emp_no", "emp_no.*"));
+    }
+
+    public void testIn2() {
+        assertFieldNames("""
+            from employees
+            | eval x = date_trunc(hire_date, 1 year)
+            | where birth_date not in (x, hire_date)
+            | keep x, hire_date
+            | sort x desc
+            | limit 4""", Set.of("hire_date", "hire_date.*", "birth_date", "birth_date.*"));
+    }
+
+    public void testAutoBucketMonth() {
+        assertFieldNames("""
+            from employees
+            | where hire_date >= "1985-01-01T00:00:00Z" and hire_date < "1986-01-01T00:00:00Z"
+            | eval hd = auto_bucket(hire_date, 20, "1985-01-01T00:00:00Z", "1986-01-01T00:00:00Z")
+            | sort hire_date
+            | keep hire_date, hd""", Set.of("hire_date", "hire_date.*"));
+    }
+
+    public void testBorn_before_today() {
+        assertFieldNames(
+            "from employees | where birth_date < now() | sort emp_no asc | keep emp_no, birth_date| limit 1",
+            Set.of("birth_date", "birth_date.*", "emp_no", "emp_no.*")
+        );
+    }
+
+    public void testAutoBucketMonthInAgg() {
+        assertFieldNames("""
+            FROM employees
+            | WHERE hire_date >= "1985-01-01T00:00:00Z" AND hire_date < "1986-01-01T00:00:00Z"
+            | EVAL bucket = AUTO_BUCKET(hire_date, 20, "1985-01-01T00:00:00Z", "1986-01-01T00:00:00Z")
+            | STATS AVG(salary) BY bucket
+            | SORT bucket""", Set.of("salary", "salary.*", "hire_date", "hire_date.*"));
+    }
+
+    public void testEvalDateParseDynamic() {
+        assertFieldNames("""
+            from employees
+            | where emp_no == 10039 or emp_no == 10040
+            | sort emp_no
+            | eval birth_date_string = date_format(birth_date, "yyyy-MM-dd")
+            | eval new_date = date_parse(birth_date_string, "yyyy-MM-dd")
+            | eval bool = new_date == birth_date
+            | keep emp_no, new_date, birth_date, bool""", Set.of("emp_no", "emp_no.*", "birth_date", "birth_date.*"));
+    }
+
+    public void testDateFields() {
+        assertFieldNames("""
+            from employees
+            | where emp_no == 10049 or emp_no == 10050
+            | eval year = date_extract(birth_date, "year"), month = date_extract(birth_date, "month_of_year")
+            | keep emp_no, year, month""", Set.of("emp_no", "emp_no.*", "birth_date", "birth_date.*"));
+    }
+
+    public void testEvalDissect() {
+        assertFieldNames("""
+            from employees
+            | eval full_name = concat(first_name, " ", last_name)
+            | dissect full_name "%{a} %{b}"
+            | sort emp_no asc
+            | keep full_name, a, b
+            | limit 3""", Set.of("first_name", "first_name.*", "last_name", "last_name.*", "emp_no", "emp_no.*"));
+    }
+
+    public void testDissectExpression() {
+        assertFieldNames("""
+            from employees
+            | dissect concat(first_name, " ", last_name) "%{a} %{b}"
+            | sort emp_no asc
+            | keep a, b
+            | limit 3""", Set.of("first_name", "first_name.*", "last_name", "last_name.*", "emp_no", "emp_no.*"));
+    }
+
+    public void testMultivalueInput1() {
+        assertFieldNames("""
+            from employees
+            | where emp_no <= 10006
+            | dissect job_positions "%{a} %{b} %{c}"
+            | sort emp_no
+            | keep emp_no, a, b, c""", Set.of("emp_no", "emp_no.*", "job_positions", "job_positions.*"));
+    }
+
+    public void testDocsDropHeight() {
+        assertFieldNames("""
+            FROM employees
+            | DROP height
+            | LIMIT 0""", Set.of("*"));
+    }
+
+    public void testDocsDropHeightWithWildcard() {
+        assertFieldNames("""
+            FROM employees
+            | DROP height*
+            | LIMIT 0""", Set.of("*"));
+    }
+
+    public void testDocsEval() {
+        assertFieldNames("""
+            FROM employees
+            | KEEP first_name, last_name, height
+            | EVAL height_feet = height * 3.281, height_cm = height * 100
+            | WHERE first_name == "Georgi"
+            | LIMIT 1""", Set.of("first_name", "first_name.*", "last_name", "last_name.*", "height", "height.*"));
+    }
+
+    public void testDocsKeepWildcard() {
+        assertFieldNames("""
+            FROM employees
+            | KEEP h*
+            | LIMIT 0""", Set.of("h*"));
+    }
+
+    public void testDocsKeepDoubleWildcard() {
+        assertFieldNames("""
+            FROM employees
+            | KEEP h*, *
+            | LIMIT 0""", Set.of("*"));
+    }
+
+    public void testDocsRename() {
+        assertFieldNames("""
+            FROM employees
+            | KEEP first_name, last_name, still_hired
+            | RENAME  still_hired AS employed
+            | LIMIT 0""", Set.of("first_name", "first_name.*", "last_name", "last_name.*", "still_hired", "still_hired.*"));
+    }
+
+    public void testDocsRenameMultipleColumns() {
+        assertFieldNames("""
+            FROM employees
+            | KEEP first_name, last_name
+            | RENAME first_name AS fn, last_name AS ln
+            | LIMIT 0""", Set.of("first_name", "first_name.*", "last_name", "last_name.*"));
+    }
+
+    public void testDocsStats() {
+        assertFieldNames("""
+            FROM employees
+            | STATS count = COUNT(emp_no) BY languages
+            | SORT languages""", Set.of("emp_no", "emp_no.*", "languages", "languages.*"));
+    }
+
+    public void testSortWithLimitOne_DropHeight() {
+        assertFieldNames("from employees | sort languages | limit 1 | drop height*", Set.of("*"));
+    }
+
+    public void testDropAllColumns() {
+        assertFieldNames("from employees | keep height | drop height | eval x = 1", Set.of("height", "height.*"));
+    }
+
+    public void testDropAllColumns_WithStats() {
+        assertFieldNames(
+            "from employees | keep height | drop height | eval x = 1 | stats c=count(x), mi=min(x), s=sum(x)",
+            Set.of("height", "height.*")
+        );
+    }
+
+    public void testEnrichOn() {
+        assertFieldNames("""
+            from employees
+            | sort emp_no
+            | limit 1
+            | eval x = to_string(languages)
+            | enrich languages_policy on x
+            | keep emp_no, language_name""", Set.of("emp_no", "emp_no.*", "languages", "languages.*", "language_name", "language_name.*"));
+    }
+
+    public void testEnrichOn2() {
+        assertFieldNames("""
+            from employees
+            | eval x = to_string(languages)
+            | enrich languages_policy on x
+            | keep emp_no, language_name
+            | sort emp_no
+            | limit 1""", Set.of("emp_no", "emp_no.*", "languages", "languages.*", "language_name", "language_name.*"));
+    }
+
+    public void testUselessEnrich() {
+        assertFieldNames("""
+            from employees
+            | eval x = "abc"
+            | enrich languages_policy on x
+            | limit 1""", Set.of("*"));
+    }
+
+    public void testSimpleSortLimit() {
+        assertFieldNames("""
+            from employees
+            | eval x = to_string(languages)
+            | enrich languages_policy on x
+            | keep emp_no, language_name
+            | sort emp_no
+            | limit 1""", Set.of("languages", "languages.*", "emp_no", "emp_no.*", "language_name", "language_name.*"));
+    }
+
+    public void testWith() {
+        assertFieldNames(
+            """
+                from employees | eval x = to_string(languages) | keep emp_no, x | sort emp_no | limit 1
+                | enrich languages_policy on x with language_name""",
+            Set.of("emp_no", "emp_no.*", "languages", "languages.*", "language_name", "language_name.*")
+        );
+    }
+
+    public void testWithAlias() {
+        assertFieldNames(
+            """
+                from employees  | sort emp_no | limit 3 | eval x = to_string(languages) | keep emp_no, x
+                | enrich languages_policy on x with lang = language_name""",
+            Set.of("emp_no", "emp_no.*", "languages", "languages.*", "language_name", "language_name.*")
+        );
+    }
+
+    public void testWithAliasSort() {
+        assertFieldNames(
+            """
+                from employees | eval x = to_string(languages) | keep emp_no, x  | sort emp_no | limit 3
+                | enrich languages_policy on x with lang = language_name""",
+            Set.of("emp_no", "emp_no.*", "languages", "languages.*", "language_name", "language_name.*")
+        );
+    }
+
+    public void testWithAliasAndPlain() {
+        assertFieldNames(
+            """
+                from employees  | sort emp_no desc | limit 3 | eval x = to_string(languages) | keep emp_no, x
+                | enrich languages_policy on x with lang = language_name, language_name""",
+            Set.of("emp_no", "emp_no.*", "languages", "languages.*", "language_name", "language_name.*")
+        );
+    }
+
+    public void testWithTwoAliasesSameProp() {
+        assertFieldNames(
+            """
+                from employees  | sort emp_no | limit 1 | eval x = to_string(languages) | keep emp_no, x
+                | enrich languages_policy on x with lang = language_name, lang2 = language_name""",
+            Set.of("emp_no", "emp_no.*", "languages", "languages.*", "language_name", "language_name.*")
+        );
+    }
+
+    public void testRedundantWith() {
+        assertFieldNames(
+            """
+                from employees  | sort emp_no | limit 1 | eval x = to_string(languages) | keep emp_no, x
+                | enrich languages_policy on x with language_name, language_name""",
+            Set.of("emp_no", "emp_no.*", "languages", "languages.*", "language_name", "language_name.*")
+        );
+    }
+
+    public void testNullInput() {
+        assertFieldNames(
+            """
+                from employees
+                | where emp_no == 10017
+                | keep emp_no, gender
+                | enrich languages_policy on gender with language_name, language_name""",
+            Set.of("gender", "gender.*", "emp_no", "emp_no.*", "language_name", "language_name.*")
+        );
+    }
+
+    public void testConstantNullInput() {
+        assertFieldNames(
+            """
+                from employees
+                | where emp_no == 10020
+                | eval x = to_string(languages)
+                | keep emp_no, x
+                | enrich languages_policy on x with language_name, language_name""",
+            Set.of("languages", "languages.*", "emp_no", "emp_no.*", "language_name", "language_name.*")
+        );
+    }
+
+    public void testEnrichEval() {
+        assertFieldNames("""
+            from employees
+            | eval x = to_string(languages)
+            | enrich languages_policy on x with lang = language_name
+            | eval language = concat(x, "-", lang)
+            | keep emp_no, x, lang, language
+            | sort emp_no desc | limit 3""", Set.of("languages", "languages.*", "emp_no", "emp_no.*", "language_name", "language_name.*"));
+    }
+
+    public void testSimple() {
+        assertFieldNames("""
+            from employees
+            | eval x = 1, y = to_string(languages)
+            | enrich languages_policy on y
+            | where x > 1
+            | keep emp_no, language_name
+            | limit 1""", Set.of("emp_no", "emp_no.*", "languages", "languages.*", "language_name", "language_name.*"));
+    }
+
+    public void testEvalNullSort() {
+        assertFieldNames(
+            "from employees | eval x = null | sort x asc, emp_no desc | keep emp_no, x, last_name | limit 2",
+            Set.of("last_name", "last_name.*", "emp_no", "emp_no.*")
+        );
+    }
+
+    public void testFilterEvalFilter() {
+        assertFieldNames("""
+            from employees
+            | where emp_no < 100010
+            | eval name_len = length(first_name)
+            | where name_len < 4
+            | keep first_name
+            | sort first_name""", Set.of("emp_no", "emp_no.*", "first_name", "first_name.*"));
+    }
+
+    public void testEvalWithIsNullIsNotNull() {
+        assertFieldNames(
+            """
+                from employees
+                | eval true_bool = null is null, false_bool = null is not null, negated_true = not(null is null)
+                | sort emp_no
+                | limit 1
+                | keep *true*, *false*, first_name, last_name""",
+            Set.of("emp_no", "emp_no.*", "first_name", "first_name.*", "last_name", "last_name.*", "*true*", "*false*")
+        );
+    }
+
+    public void testInDouble() {
+        assertFieldNames(
+            "from employees | keep emp_no, height, height.float, height.half_float, height.scaled_float | where height in (2.03)",
+            Set.of(
+                "emp_no",
+                "emp_no.*",
+                "height",
+                "height.*",
+                "height.float",
+                "height.float.*",
+                "height.half_float",
+                "height.half_float.*",
+                "height.scaled_float",
+                "height.scaled_float.*"
+            )
+        );
+    }
+
+    public void testConvertFromDatetime() {
+        assertFieldNames(
+            "from employees | sort emp_no | eval hire_double = to_double(hire_date) | keep emp_no, hire_date, hire_double | limit 3",
+            Set.of("emp_no", "emp_no.*", "hire_date", "hire_date.*")
+        );
+    }
+
+    public void testAutoBucket() {
+        assertFieldNames("""
+            FROM employees
+            | WHERE hire_date >= "1985-01-01T00:00:00Z" AND hire_date < "1986-01-01T00:00:00Z"
+            | EVAL bh = auto_bucket(height, 20, 1.41, 2.10)
+            | SORT hire_date
+            | KEEP hire_date, height, bh""", Set.of("hire_date", "hire_date.*", "height", "height.*"));
+    }
+
+    public void testEvalGrok() {
+        assertFieldNames("""
+            from employees
+            | eval full_name = concat(first_name, " ", last_name)
+            | grok full_name "%{WORD:a} %{WORD:b}"
+            | sort emp_no asc
+            | keep full_name, a, b
+            | limit 3""", Set.of("first_name", "first_name.*", "last_name", "last_name.*", "emp_no", "emp_no.*"));
+    }
+
+    public void testGrokExpression() {
+        assertFieldNames("""
+            from employees
+            | grok concat(first_name, " ", last_name) "%{WORD:a} %{WORD:b}"
+            | sort emp_no asc
+            | keep a, b
+            | limit 3""", Set.of("first_name", "first_name.*", "last_name", "last_name.*", "emp_no", "emp_no.*"));
+    }
+
+    public void testEvalGrokSort() {
+        assertFieldNames("""
+            from employees
+            | eval full_name = concat(first_name, " ", last_name)
+            | grok full_name "%{WORD:a} %{WORD:b}"
+            | sort a asc
+            | keep full_name, a, b
+            | limit 3""", Set.of("first_name", "first_name.*", "last_name", "last_name.*"));
+    }
+
+    public void testGrokStats() {
+        assertFieldNames("""
+            from employees
+            | eval x = concat(gender, " foobar")
+            | grok x "%{WORD:a} %{WORD:b}"
+            | stats n = max(emp_no) by a
+            | keep a, n
+            | sort a asc""", Set.of("gender", "gender.*", "emp_no", "emp_no.*"));
+    }
+
+    public void testNullOnePattern() {
+        assertFieldNames("""
+            from employees
+            | where emp_no == 10030
+            | grok first_name "%{WORD:a}"
+            | keep first_name, a""", Set.of("first_name", "first_name.*", "emp_no", "emp_no.*"));
+    }
+
+    public void testMultivalueInput() {
+        assertFieldNames("""
+            from employees
+            | where emp_no <= 10006
+            | grok job_positions "%{WORD:a} %{WORD:b} %{WORD:c}"
+            | sort emp_no
+            | keep emp_no, a, b, c, job_positions""", Set.of("job_positions", "job_positions.*", "emp_no", "emp_no.*"));
+    }
+
+    public void testSelectAll() {
+        assertFieldNames("FROM apps [metadata _id]", Set.of("*"));
+    }
+
+    public void testFilterById() {
+        assertFieldNames("FROM apps [metadata _id]| WHERE _id == \"4\"", Set.of("*"));
+    }
+
+    public void testKeepId() {
+        assertFieldNames("FROM apps [metadata _id] | WHERE id == 3 | KEEP _id", Set.of("id", "id.*"));
+    }
+
+    public void testIdRangeAndSort() {
+        assertFieldNames("""
+            FROM apps [metadata _id]
+            | WHERE _id >= "2" AND _id <= "7"
+            | SORT _id
+            | keep id, name, _id""", Set.of("id", "id.*", "name", "name.*"));
+    }
+
+    public void testOrderById() {
+        assertFieldNames("FROM apps [metadata _id] | KEEP _id, name | SORT _id", Set.of("name", "name.*"));
+    }
+
+    public void testOrderByIdDesc() {
+        assertFieldNames("FROM apps [metadata _id] | KEEP _id, name | SORT _id DESC", Set.of("name", "name.*"));
+    }
+
+    public void testConcatId() {
+        assertFieldNames("FROM apps [metadata _id] | eval c = concat(_id, name) | SORT _id | KEEP c", Set.of("name", "name.*"));
+    }
+
+    public void testStatsOnId() {
+        assertFieldNames("FROM apps [metadata _id] | stats c = count(_id), d = count_distinct(_id)", Set.of("*"));
+    }
+
+    public void testStatsOnIdByGroup() {
+        assertFieldNames("FROM apps [metadata _id] | stats c = count(_id) by name | sort c desc, name | limit 5", Set.of("name", "name.*"));
+    }
+
+    public void testSimpleProject() {
+        assertFieldNames(
+            "from hosts | keep card, host, ip0, ip1",
+            Set.of("card", "card.*", "host", "host.*", "ip0", "ip0.*", "ip1", "ip1.*")
+        );
+    }
+
+    public void testEquals() {
+        assertFieldNames(
+            "from hosts | sort host, card | where ip0 == ip1 | keep card, host",
+            Set.of("card", "card.*", "host", "host.*", "ip0", "ip0.*", "ip1", "ip1.*")
+        );
+    }
+
+    public void testConditional() {
+        assertFieldNames("from hosts | eval eq=case(ip0==ip1, ip0, ip1) | keep eq, ip0, ip1", Set.of("ip1", "ip1.*", "ip0", "ip0.*"));
+    }
+
+    public void testWhereWithAverageBySubField() {
+        assertFieldNames(
+            "from employees | where languages + 1 == 6 | stats avg(avg_worked_seconds) by languages.long",
+            Set.of("languages", "languages.*", "avg_worked_seconds", "avg_worked_seconds.*", "languages.long", "languages.long.*")
+        );
+    }
+
+    public void testAverageOfEvalValue() {
+        assertFieldNames(
+            "from employees | eval ratio = salary / height | stats avg(ratio)",
+            Set.of("salary", "salary.*", "height", "height.*")
+        );
+    }
+
+    public void testTopNProjectEvalProject() {
+        assertFieldNames(
+            "from employees | sort salary | limit 1 | keep languages, salary | eval x = languages + 1 | keep x",
+            Set.of("salary", "salary.*", "languages", "languages.*")
+        );
+    }
+
+    public void testMvSum() {
+        assertFieldNames("""
+            from employees
+            | where emp_no > 10008
+            | eval salary_change = mv_sum(salary_change.int)
+            | sort emp_no
+            | keep emp_no, salary_change.int, salary_change
+            | limit 7""", Set.of("emp_no", "emp_no.*", "salary_change.int", "salary_change.int.*"));
+    }
+
+    public void testMetaIndexAliasedInAggs() {
+        assertFieldNames(
+            "from employees [metadata _index] | eval _i = _index | stats max = max(emp_no) by _i",
+            Set.of("emp_no", "emp_no.*")
+        );
+    }
+
+    public void testCoalesceFolding() {
+        assertFieldNames("""
+            FROM employees
+            | EVAL foo=COALESCE(true, false, null)
+            | SORT emp_no ASC
+            | KEEP emp_no, first_name, foo
+            | limit 3""", Set.of("emp_no", "emp_no.*", "first_name", "first_name.*"));
+    }
+
+    public void testRenameEvalProject() {
+        assertFieldNames(
+            "from employees | rename languages as x | keep x | eval z = 2 * x | keep x, z | limit 3",
+            Set.of("languages", "languages.*")
+        );
+    }
+
+    public void testRenameProjectEval() {
+        assertFieldNames("""
+            from employees
+            | eval y = languages
+            | rename languages as x
+            | keep x, y
+            | eval x2 = x + 1
+            | eval y2 = y + 2
+            | limit 3""", Set.of("languages", "languages.*"));
+    }
+
+    public void testRenameWithFilterPushedToES() {
+        assertFieldNames(
+            "from employees | rename emp_no as x | keep languages, first_name, last_name, x | where x > 10030 and x < 10040 | limit 5",
+            Set.of("emp_no", "emp_no.*", "languages", "languages.*", "first_name", "first_name.*", "last_name", "last_name.*")
+        );
+    }
+
+    public void testRenameOverride() {
+        assertFieldNames(
+            "from employees | rename emp_no as languages | keep languages, last_name | limit 3",
+            Set.of("emp_no", "emp_no.*", "last_name", "last_name.*")
+        );
+    }
+
+    public void testProjectRenameDate() {
+        assertFieldNames(
+            "from employees | sort hire_date | rename hire_date as x | keep emp_no, x | limit 5",
+            Set.of("hire_date", "hire_date.*", "emp_no", "emp_no.*")
+        );
+    }
+
+    public void testRenameDrop() {
+        assertFieldNames("""
+            from employees
+            | sort hire_date
+            | rename hire_date as x, emp_no as y
+            | drop first_name, last_name, gender, birth_date, salary, languages*, height*, still_hired, avg_worked_seconds,
+            job_positions, is_rehired, salary_change*
+            | limit 5""", Set.of("*"));
+    }
+
+    public void testMaxOfLong() {
+        assertFieldNames("from employees | stats l = max(languages.long)", Set.of("languages.long", "languages.long.*"));
+    }
+
+    public void testGroupByAlias() {
+        assertFieldNames(
+            "from employees | rename languages as l | keep l, height | stats m = min(height) by l | sort l",
+            Set.of("languages", "languages.*", "height", "height.*")
+        );
+    }
+
+    public void testByStringAndLong() {
+        assertFieldNames("""
+            from employees
+            | eval trunk_worked_seconds = avg_worked_seconds / 100000000 * 100000000
+            | stats c = count(gender) by gender, trunk_worked_seconds
+            | sort c desc""", Set.of("avg_worked_seconds", "avg_worked_seconds.*", "gender", "gender.*"));
+    }
+
+    public void testByStringAndLongWithAlias() {
+        assertFieldNames("""
+            from employees
+            | eval trunk_worked_seconds = avg_worked_seconds / 100000000 * 100000000
+            | rename gender as g, trunk_worked_seconds as tws
+            | keep g, tws
+            | stats c = count(g) by g, tws
+            | sort c desc""", Set.of("avg_worked_seconds", "avg_worked_seconds.*", "gender", "gender.*"));
+    }
+
+    public void testByStringAndString() {
+        assertFieldNames("""
+            from employees
+            | eval hire_year_str = date_format(hire_date, "yyyy")
+            | stats c = count(gender) by gender, hire_year_str
+            | sort c desc, gender, hire_year_str
+            | where c >= 5""", Set.of("hire_date", "hire_date.*", "gender", "gender.*"));
+    }
+
+    public void testByLongAndLong() {
+        assertFieldNames("""
+            from employees
+            | eval trunk_worked_seconds = avg_worked_seconds / 100000000 * 100000000
+            | stats c = count(languages.long) by languages.long, trunk_worked_seconds
+            | sort c desc""", Set.of("avg_worked_seconds", "avg_worked_seconds.*", "languages.long", "languages.long.*"));
+    }
+
+    public void testByDateAndKeywordAndIntWithAlias() {
+        assertFieldNames("""
+            from employees
+            | eval d = date_trunc(hire_date, 1 year)
+            | rename gender as g, languages as l, emp_no as e
+            | keep d, g, l, e
+            | stats c = count(e) by d, g, l
+            | sort c desc, d, l desc
+            | limit 10""", Set.of("hire_date", "hire_date.*", "gender", "gender.*", "languages", "languages.*", "emp_no", "emp_no.*"));
+    }
+
+    public void testCountDistinctOfKeywords() {
+        assertFieldNames(
+            """
+                from employees
+                | eval hire_year_str = date_format(hire_date, "yyyy")
+                | stats g = count_distinct(gender), h = count_distinct(hire_year_str)""",
+            Set.of("hire_date", "hire_date.*", "gender", "gender.*")
+        );
+    }
+
+    public void testCountDistinctOfIpPrecision() {
+        assertFieldNames("""
+            FROM hosts
+            | STATS COUNT_DISTINCT(ip0, 80000), COUNT_DISTINCT(ip1, 5)""", Set.of("ip0", "ip0.*", "ip1", "ip1.*"));
+    }
+
+    public void testPercentileOfLong() {
+        assertFieldNames(
+            """
+                from employees
+                | stats p0 = percentile(salary_change.long, 0), p50 = percentile(salary_change.long, 50)""",
+            Set.of("salary_change.long", "salary_change.long.*")
+        );
+    }
+
+    public void testMedianOfInteger() {
+        assertFieldNames("""
+            FROM employees
+            | STATS MEDIAN(salary), PERCENTILE(salary, 50)""", Set.of("salary", "salary.*"));
+    }
+
+    public void testMedianAbsoluteDeviation() {
+        assertFieldNames("""
+            FROM employees
+            | STATS MEDIAN(salary), MEDIAN_ABSOLUTE_DEVIATION(salary)""", Set.of("salary", "salary.*"));
+    }
+
+    public void testIn3VLWithComputedNull() {
+        assertFieldNames(
+            """
+                from employees
+                | where mv_count(job_positions) <= 1
+                | where emp_no >= 10024
+                | limit 3
+                | keep emp_no, job_positions
+                | eval nil = concat("", null)
+                | eval is_in = job_positions in ("Accountant", "Internship", nil)""",
+            Set.of("job_positions", "job_positions.*", "emp_no", "emp_no.*")
+        );
+    }
+
+    public void testCase() {
+        assertFieldNames("""
+            FROM apps
+            | EVAL version_text = TO_STR(version)
+            | WHERE version IS NULL OR version_text LIKE "1*"
+            | EVAL v = TO_VER(CONCAT("123", TO_STR(version)))
+            | EVAL m = CASE(version > TO_VER("1.1"), 1, 0)
+            | EVAL g = CASE(version > TO_VER("1.3.0"), version, TO_VER("1.3.0"))
+            | EVAL i = CASE(version IS NULL, TO_VER("0.1"), version)
+            | EVAL c = CASE(
+            version > TO_VER("1.1"), "high",
+            version IS NULL, "none",
+            "low")
+            | SORT version DESC NULLS LAST, id DESC
+            | KEEP v, version, version_text, id, m, g, i, c""", Set.of("version", "version.*", "id", "id.*"));
+    }
+
+    public void testLikePrefix() {
+        assertFieldNames("""
+            from employees
+            | where first_name like "Eberhar*"
+            | keep emp_no, first_name""", Set.of("emp_no", "emp_no.*", "first_name", "first_name.*"));
+    }
+
+    public void testRLikePrefix() {
+        assertFieldNames("""
+            from employees
+            | where first_name rlike "Aleja.*"
+            | keep emp_no""", Set.of("first_name", "first_name.*", "emp_no", "emp_no.*"));
+    }
+
+    public void testByUnmentionedLongAndLong() {
+        assertFieldNames(
+            """
+                from employees
+                | eval trunk_worked_seconds = avg_worked_seconds / 100000000 * 100000000
+                | stats c = count(gender) by languages.long, trunk_worked_seconds
+                | sort c desc""",
+            Set.of("avg_worked_seconds", "avg_worked_seconds.*", "languages.long", "languages.long.*", "gender", "gender.*")
+        );
+    }
+
+    public void testRenameNopProject() {
+        assertFieldNames("""
+            from employees
+            | rename emp_no as emp_no
+            | keep emp_no, last_name
+            | limit 3""", Set.of("emp_no", "emp_no.*", "last_name", "last_name.*"));
+    }
+
+    public void testRename() {
+        assertFieldNames("""
+            from test
+            | rename emp_no as e
+            | keep first_name, e
+            """, Set.of("emp_no", "emp_no.*", "first_name", "first_name.*"));
+    }
+
+    public void testChainedRename() {
+        assertFieldNames("""
+            from test
+            | rename emp_no as r1, r1 as r2, r2 as r3
+            | keep first_name, r3
+            """, Set.of("emp_no", "emp_no.*", "first_name", "first_name.*", "r1", "r1.*", "r2", "r2.*"));// TODO asking for more shouldn't
+                                                                                                         // hurt. Can we do better?
+        // Set.of("emp_no", "emp_no.*", "first_name", "first_name.*"));
+    }
+
+    public void testChainedRenameReuse() {
+        assertFieldNames("""
+            from test
+            | rename emp_no as r1, r1 as r2, r2 as r3, first_name as r1
+            | keep r1, r3
+            """, Set.of("emp_no", "emp_no.*", "first_name", "first_name.*", "r1", "r1.*", "r2", "r2.*"));// TODO asking for more shouldn't
+                                                                                                         // hurt. Can we do better?
+        // Set.of("emp_no", "emp_no.*", "first_name", "first_name.*"));
+    }
+
+    public void testRenameBackAndForth() {
+        assertFieldNames("""
+            from test
+            | rename emp_no as r1, r1 as emp_no
+            | keep emp_no
+            """, Set.of("emp_no", "emp_no.*", "r1", "r1.*"));// TODO asking for more shouldn't hurt. Can we do better?
+        // Set.of("emp_no", "emp_no.*"));
+    }
+
+    public void testRenameReuseAlias() {
+        assertFieldNames("""
+            from test
+            | rename emp_no as e, first_name as e
+            """, Set.of("*"));
+    }
+
+    public void testIfDuplicateNamesGroupingHasPriority() {
+        assertFieldNames(
+            "from employees | stats languages = avg(height), languages = min(height) by languages | sort languages",
+            Set.of("height", "height.*", "languages", "languages.*")
+        );
+    }
+
+    public void testCoalesce() {
+        assertFieldNames("""
+            FROM employees
+            | EVAL first_name = COALESCE(first_name, "X")
+            | SORT first_name DESC, emp_no ASC
+            | KEEP emp_no, first_name
+            | limit 10""", Set.of("first_name", "first_name.*", "emp_no", "emp_no.*"));
+    }
+
+    public void testCoalesceBackwards() {
+        assertFieldNames("""
+            FROM employees
+            | EVAL first_name = COALESCE("X", first_name)
+            | SORT first_name DESC, emp_no ASC
+            | KEEP emp_no, first_name
+            | limit 10""", Set.of("first_name", "first_name.*", "emp_no", "emp_no.*"));
+    }
+
+    public void testGroupByVersionCast() {
+        assertFieldNames("""
+            FROM apps
+            | EVAL g = TO_VER(CONCAT("1.", TO_STR(version)))
+            | STATS id = MAX(id) BY g
+            | SORT id
+            | DROP g""", Set.of("version", "version.*", "id", "id.*"));
+    }
+
+    public void testCoalesceEndsInNull() {
+        assertFieldNames("""
+            FROM employees
+            | EVAL first_name = COALESCE(first_name, last_name, null)
+            | SORT first_name DESC, emp_no ASC
+            | KEEP emp_no, first_name
+            | limit 3""", Set.of("first_name", "first_name.*", "last_name", "last_name.*", "emp_no", "emp_no.*"));
+    }
+
+    public void testMvAvg() {
+        assertFieldNames("""
+            from employees
+            | where emp_no > 10008
+            | eval salary_change = mv_avg(salary_change)
+            | sort emp_no
+            | keep emp_no, salary_change.int, salary_change
+            | limit 7""", Set.of("emp_no", "emp_no.*", "salary_change", "salary_change.*", "salary_change.int", "salary_change.int.*"));
+    }
+
+    public void testEvalOverride() {
+        assertFieldNames("""
+            from employees
+            | eval languages = languages + 1
+            | eval languages = languages + 1
+            | limit 5
+            | keep l*""", Set.of("languages", "languages.*", "l*"));// subtlety here. Keeping only "languages*" can remove any other "l*"
+                                                                    // named fields
+    }
+
+    public void testBasicWildcardKeep() {
+        assertFieldNames("from test | keep *", Set.of("*"));
+    }
+
+    public void testBasicWildcardKeep2() {
+        assertFieldNames("""
+            from test
+            | keep un*
+            """, Set.of("un*"));
+    }
+
+    public void testWildcardKeep() {
+        assertFieldNames("""
+            from test
+            | keep first_name, *, last_name
+            """, Set.of("*"));
+    }
+
+    public void testProjectThenDropName() {
+        assertFieldNames("""
+            from test
+            | keep *name
+            | drop first_name
+            """, Set.of("*name", "*name.*", "first_name", "first_name.*"));
+    }
+
+    public void testProjectAfterDropName() {
+        assertFieldNames("""
+            from test
+            | drop first_name
+            | keep *name
+            """, Set.of("*name.*", "*name", "first_name", "first_name.*"));
+    }
+
+    public void testProjectKeepAndDropName() {
+        assertFieldNames("""
+            from test
+            | drop first_name
+            | keep last_name
+            """, Set.of("last_name", "last_name.*", "first_name", "first_name.*"));
+    }
+
+    public void testProjectDropPattern() {
+        assertFieldNames("""
+            from test
+            | keep *
+            | drop *_name
+            """, Set.of("*"));
+    }
+
+    public void testProjectDropNoStarPattern() {
+        assertFieldNames("""
+            from test
+            | drop *_name
+            """, Set.of("*"));
+    }
+
+    public void testProjectOrderPatternWithRest() {
+        assertFieldNames("""
+            from test
+            | keep *name, *, emp_no
+            """, Set.of("*"));
+    }
+
+    public void testProjectDropPatternAndKeepOthers() {
+        assertFieldNames("""
+            from test
+            | drop l*
+            | keep first_name, salary
+            """, Set.of("l*", "first_name", "first_name.*", "salary", "salary.*"));
+    }
+
+    public void testAliasesThatGetDropped() {
+        assertFieldNames("""
+            from test
+            | eval x = languages + 1
+            | where first_name like "%A"
+            | eval first_name = concat(first_name, "xyz")
+            | drop first_name
+            """, Set.of("*"));
+    }
+
+    private void assertFieldNames(String query, Set<String> expected) {
+        Set<String> fieldNames = EsqlSession.fieldNames(parser.createStatement(query));
+        assertThat(fieldNames, equalTo(expected));
+    }
+}

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/index/IndexResolver.java

@@ -173,6 +173,7 @@ public class IndexResolver {
         EnumSet.of(WildcardStates.OPEN)
     );
 
+    public static final Set<String> ALL_FIELDS = Set.of("*");
     private static final String UNMAPPED = "unmapped";
 
     private final Client client;
@@ -304,6 +305,7 @@ public class IndexResolver {
             IndicesOptions indicesOptions = retrieveFrozenIndices ? FIELD_CAPS_FROZEN_INDICES_OPTIONS : FIELD_CAPS_INDICES_OPTIONS;
             FieldCapabilitiesRequest fieldRequest = createFieldCapsRequest(
                 qualifyAndJoinIndices(clusterWildcard, indexWildcards),
+                ALL_FIELDS,
                 indicesOptions,
                 emptyMap()
             );
@@ -356,11 +358,12 @@ public class IndexResolver {
      */
     public void resolveAsMergedMapping(
         String indexWildcard,
+        Set<String> fieldNames,
         IndicesOptions indicesOptions,
         Map<String, Object> runtimeMappings,
         ActionListener<IndexResolution> listener
     ) {
-        FieldCapabilitiesRequest fieldRequest = createFieldCapsRequest(indexWildcard, indicesOptions, runtimeMappings);
+        FieldCapabilitiesRequest fieldRequest = createFieldCapsRequest(indexWildcard, fieldNames, indicesOptions, runtimeMappings);
         client.fieldCaps(
             fieldRequest,
             listener.delegateFailureAndWrap((l, response) -> l.onResponse(mergedMappings(typeRegistry, indexWildcard, response)))
@@ -372,11 +375,12 @@ public class IndexResolver {
      */
     public void resolveAsMergedMapping(
         String indexWildcard,
+        Set<String> fieldNames,
         boolean includeFrozen,
         Map<String, Object> runtimeMappings,
         ActionListener<IndexResolution> listener
     ) {
-        FieldCapabilitiesRequest fieldRequest = createFieldCapsRequest(indexWildcard, includeFrozen, runtimeMappings);
+        FieldCapabilitiesRequest fieldRequest = createFieldCapsRequest(indexWildcard, fieldNames, includeFrozen, runtimeMappings);
         client.fieldCaps(
             fieldRequest,
             listener.delegateFailureAndWrap((l, response) -> l.onResponse(mergedMappings(typeRegistry, indexWildcard, response)))
@@ -559,11 +563,12 @@ public class IndexResolver {
 
     private static FieldCapabilitiesRequest createFieldCapsRequest(
         String index,
+        Set<String> fieldNames,
         IndicesOptions indicesOptions,
         Map<String, Object> runtimeMappings
     ) {
         return new FieldCapabilitiesRequest().indices(Strings.commaDelimitedListToStringArray(index))
-            .fields("*")
+            .fields(fieldNames.toArray(String[]::new))
             .includeUnmapped(true)
             .runtimeFields(runtimeMappings)
             // lenient because we throw our own errors looking at the response e.g. if something was not resolved
@@ -573,11 +578,12 @@ public class IndexResolver {
 
     private static FieldCapabilitiesRequest createFieldCapsRequest(
         String index,
+        Set<String> fieldNames,
         boolean includeFrozen,
         Map<String, Object> runtimeMappings
     ) {
         IndicesOptions indicesOptions = includeFrozen ? FIELD_CAPS_FROZEN_INDICES_OPTIONS : FIELD_CAPS_INDICES_OPTIONS;
-        return createFieldCapsRequest(index, indicesOptions, runtimeMappings);
+        return createFieldCapsRequest(index, fieldNames, indicesOptions, runtimeMappings);
     }
 
     /**
@@ -590,7 +596,7 @@ public class IndexResolver {
         Map<String, Object> runtimeMappings,
         ActionListener<List<EsIndex>> listener
     ) {
-        FieldCapabilitiesRequest fieldRequest = createFieldCapsRequest(indexWildcard, includeFrozen, runtimeMappings);
+        FieldCapabilitiesRequest fieldRequest = createFieldCapsRequest(indexWildcard, ALL_FIELDS, includeFrozen, runtimeMappings);
         client.fieldCaps(fieldRequest, listener.delegateFailureAndWrap((delegate, response) -> {
             client.admin().indices().getAliases(createGetAliasesRequest(response, includeFrozen), wrap(aliases -> {
                 delegate.onResponse(separateMappings(typeRegistry, javaRegex, response, aliases.getAliases()));

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/plan/logical/command/ShowColumns.java

@@ -12,6 +12,7 @@ import org.elasticsearch.xpack.ql.expression.Attribute;
 import org.elasticsearch.xpack.ql.expression.FieldAttribute;
 import org.elasticsearch.xpack.ql.expression.predicate.regex.LikePattern;
 import org.elasticsearch.xpack.ql.index.IndexCompatibility;
+import org.elasticsearch.xpack.ql.index.IndexResolver;
 import org.elasticsearch.xpack.ql.tree.NodeInfo;
 import org.elasticsearch.xpack.ql.tree.Source;
 import org.elasticsearch.xpack.ql.type.DataType;
@@ -80,15 +81,22 @@ public class ShowColumns extends Command {
         idx = hasText(cat) && cat.equals(cluster) == false ? buildRemoteIndexName(cat, idx) : idx;
 
         boolean withFrozen = includeFrozen || session.configuration().includeFrozen();
-        session.indexResolver().resolveAsMergedMapping(idx, withFrozen, emptyMap(), listener.delegateFailureAndWrap((l, indexResult) -> {
-            List<List<?>> rows = emptyList();
-            if (indexResult.isValid()) {
-                rows = new ArrayList<>();
-                Version version = Version.fromId(session.configuration().version().id);
-                fillInRows(IndexCompatibility.compatible(indexResult, version).get().mapping(), null, rows);
-            }
-            l.onResponse(of(session, rows));
-        }));
+        session.indexResolver()
+            .resolveAsMergedMapping(
+                idx,
+                IndexResolver.ALL_FIELDS,
+                withFrozen,
+                emptyMap(),
+                listener.delegateFailureAndWrap((l, indexResult) -> {
+                    List<List<?>> rows = emptyList();
+                    if (indexResult.isValid()) {
+                        rows = new ArrayList<>();
+                        Version version = Version.fromId(session.configuration().version().id);
+                        fillInRows(IndexCompatibility.compatible(indexResult, version).get().mapping(), null, rows);
+                    }
+                    l.onResponse(of(session, rows));
+                })
+            );
     }
 
     static void fillInRows(Map<String, EsField> mapping, String prefix, List<List<?>> rows) {

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

@@ -14,6 +14,7 @@ import org.elasticsearch.xpack.ql.expression.Attribute;
 import org.elasticsearch.xpack.ql.expression.predicate.regex.LikePattern;
 import org.elasticsearch.xpack.ql.index.EsIndex;
 import org.elasticsearch.xpack.ql.index.IndexCompatibility;
+import org.elasticsearch.xpack.ql.index.IndexResolver;
 import org.elasticsearch.xpack.ql.tree.NodeInfo;
 import org.elasticsearch.xpack.ql.tree.Source;
 import org.elasticsearch.xpack.ql.type.DataType;
@@ -177,22 +178,28 @@ public class SysColumns extends Command {
         // otherwise use a merged mapping
         else {
             session.indexResolver()
-                .resolveAsMergedMapping(indexPattern, includeFrozen, emptyMap(), listener.delegateFailureAndWrap((delegate, r) -> {
-                    List<List<?>> rows = new ArrayList<>();
-                    // populate the data only when a target is found
-                    if (r.isValid()) {
-                        fillInRows(
-                            tableCat,
-                            indexName,
-                            IndexCompatibility.compatible(r, version).get().mapping(),
-                            null,
-                            rows,
-                            columnMatcher,
-                            mode
-                        );
-                    }
-                    delegate.onResponse(ListCursor.of(Rows.schema(output), rows, session.configuration().pageSize()));
-                }));
+                .resolveAsMergedMapping(
+                    indexPattern,
+                    IndexResolver.ALL_FIELDS,
+                    includeFrozen,
+                    emptyMap(),
+                    listener.delegateFailureAndWrap((delegate, r) -> {
+                        List<List<?>> rows = new ArrayList<>();
+                        // populate the data only when a target is found
+                        if (r.isValid()) {
+                            fillInRows(
+                                tableCat,
+                                indexName,
+                                IndexCompatibility.compatible(r, version).get().mapping(),
+                                null,
+                                rows,
+                                columnMatcher,
+                                mode
+                            );
+                        }
+                        delegate.onResponse(ListCursor.of(Rows.schema(output), rows, session.configuration().pageSize()));
+                    })
+                );
         }
     }
 

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

@@ -164,6 +164,7 @@ public class SqlSession implements Session {
             boolean includeFrozen = configuration.includeFrozen() || tableInfo.isFrozen();
             indexResolver.resolveAsMergedMapping(
                 indexPattern,
+                IndexResolver.ALL_FIELDS,
                 includeFrozen,
                 configuration.runtimeMappings(),
                 listener.delegateFailureAndWrap((l, indexResult) -> l.onResponse(action.apply(indexResult)))

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

@@ -52,6 +52,7 @@ import static org.elasticsearch.xpack.sql.proto.Mode.isDriver;
 import static org.elasticsearch.xpack.sql.types.SqlTypesTests.loadMapping;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -397,9 +398,9 @@ public class SysColumnsTests extends ESTestCase {
         when(resolver.clusterName()).thenReturn(CLUSTER_NAME);
         when(resolver.remoteClusters()).thenReturn(Set.of(CLUSTER_NAME));
         doAnswer(invocation -> {
-            ((ActionListener<IndexResolution>) invocation.getArguments()[3]).onResponse(IndexResolution.valid(test));
+            ((ActionListener<IndexResolution>) invocation.getArguments()[4]).onResponse(IndexResolution.valid(test));
             return Void.TYPE;
-        }).when(resolver).resolveAsMergedMapping(any(), anyBoolean(), any(), any());
+        }).when(resolver).resolveAsMergedMapping(any(), eq(IndexResolver.ALL_FIELDS), anyBoolean(), any(), any());
         doAnswer(invocation -> {
             ((ActionListener<List<EsIndex>>) invocation.getArguments()[4]).onResponse(singletonList(test));
             return Void.TYPE;