ESQL: make unmapped_fields="load" automatically trigger for partially mapped KEYWORD fields, and allow projecting partially mapped non-KEYWORD fields

x-pack/plugin/esql/qa/server/src/main/java/org/elasticsearch/xpack/esql/qa/rest/generative/GenerativeForkRestTest.java

@@ -18,7 +18,7 @@
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.APPROXIMATION_V5;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.FORK_V9;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.METRICS_GROUP_BY_ALL;
-import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.OPTIONAL_FIELDS_V2;
+import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.OPTIONAL_FIELDS_V4;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.SUBQUERY_IN_FROM_COMMAND;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.UNMAPPED_FIELDS;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.VIEWS_WITH_BRANCHING;
@@ -66,7 +66,7 @@ protected void shouldSkipTest(String testName) throws IOException {
         // FORK is not supported with unmapped_fields="load", see https://github.com/elastic/elasticsearch/issues/142033
         assumeFalse(
             "FORK is not supported with unmapped_fields=\"load\"",
-            testCase.requiredCapabilities.contains(OPTIONAL_FIELDS_V2.capabilityName())
+            testCase.requiredCapabilities.contains(OPTIONAL_FIELDS_V4.capabilityName())
         );
 
         assumeFalse(

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

@@ -417,7 +417,8 @@ public void close() {
                         for (int i = 0; i < entries.length; i++) {
                             String[] header = entries[i].split(":");
                             String name = header[0].trim();
-                            String typeName = (typeMapping != null && typeMapping.containsKey(name)) ? typeMapping.get(name)
+                            String typeName = (typeMapping != null && typeMapping.containsKey(name) && typeMapping.get(name) != null)
+                                ? typeMapping.get(name)
                                 : header.length > 1 ? header[1].trim()
                                 : null;
 

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

@@ -43,6 +43,7 @@
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
@@ -88,6 +89,17 @@ public class CsvTestsDataLoader {
         new TestDataset("employees", "mapping-default.json", "employees.csv").noSubfields(),
         new TestDataset("voyager", "mapping-voyager.json", "voyager.csv").noSubfields(),
         new TestDataset("employees_incompatible", "mapping-default-incompatible.json", "employees_incompatible.csv").noSubfields(),
+        new TestDataset("employees", "mapping-default.json", "employees.csv").withIndex("employees_no_names")
+            .withTypeMapping(removeFields("first_name", "last_name"))
+            .withDynamic("false")
+            .noSubfields(),
+        new TestDataset("employees", "mapping-default.json", "employees.csv").withIndex("employees_gender_text")
+            .withTypeMapping(Map.of("gender", "text"))
+            .noSubfields(),
+        new TestDataset("employees", "mapping-default.json", "employees.csv").withIndex("employees_no_gender")
+            .withTypeMapping(removeFields("gender"))
+            .withDynamic("false")
+            .noSubfields(),
         new TestDataset("all_types", "mapping-all-types.json", "all-types.csv"),
         new TestDataset("hosts"),
         new TestDataset("apps"),
@@ -108,6 +120,8 @@ public class CsvTestsDataLoader {
         new TestDataset("ul_logs"),
         new TestDataset("sample_data"),
         new TestDataset("partial_mapping_sample_data"),
+        // TODO: withTypeMapping should be redundant here, mapping-no_mapping_sample_data.json doesn't provide any mapping where types could
+        // be overridden
         new TestDataset("no_mapping_sample_data", "mapping-no_mapping_sample_data.json", "partial_mapping_sample_data.csv").withTypeMapping(
             Stream.of("timestamp", "client_ip", "event_duration").collect(toMap(k -> k, k -> "keyword"))
         ),
@@ -767,25 +781,52 @@ private static void load(RestClient client, TestDataset dataset, IndexCreator in
         }
     }
 
+    /**
+     * Creates a type mapping that removes the given fields from the mapping.
+     * For use with {@link TestDataset#withTypeMapping}.
+     */
+    private static Map<String, String> removeFields(String... fields) {
+        var map = new HashMap<String, String>();
+        for (String field : fields) {
+            map.put(field, null);
+        }
+        return map;
+    }
+
     public static String readMappingFile(TestDataset dataset) throws IOException {
         String mappingJsonText = dataset.loadMappings();
-        if (dataset.typeMapping == null || dataset.typeMapping.isEmpty()) {
+        boolean hasTypeMappingOverrides = dataset.typeMapping != null && dataset.typeMapping.isEmpty() == false;
+        if (hasTypeMappingOverrides == false && dataset.dynamic == null) {
             return mappingJsonText;
         }
         boolean modified = false;
         ObjectMapper mapper = new ObjectMapper();
         JsonNode mappingNode = mapper.readTree(mappingJsonText);
-        JsonNode propertiesNode = mappingNode.path("properties");
-
-        for (Map.Entry<String, String> entry : dataset.typeMapping.entrySet()) {
-            String key = entry.getKey();
-            String newType = entry.getValue();
 
-            if (propertiesNode.has(key)) {
-                modified = true;
-                ((ObjectNode) propertiesNode.get(key)).put("type", newType);
+        if (hasTypeMappingOverrides) {
+            ObjectNode propertiesNode = (ObjectNode) mappingNode.path("properties");
+            for (Map.Entry<String, String> entry : dataset.typeMapping.entrySet()) {
+                String key = entry.getKey();
+                String newType = entry.getValue();
+
+                if (newType == null) {
+                    // null value means remove the field from the mapping
+                    if (propertiesNode.has(key)) {
+                        propertiesNode.remove(key);
+                        modified = true;
+                    }
+                } else if (propertiesNode.has(key)) {
+                    ((ObjectNode) propertiesNode.get(key)).put("type", newType);
+                    modified = true;
+                }
             }
         }
+
+        if (dataset.dynamic != null) {
+            ((ObjectNode) mappingNode).put("dynamic", dataset.dynamic);
+            modified = true;
+        }
+
         if (modified) {
             return mapper.writerWithDefaultPrettyPrinter().writeValueAsString(mappingNode);
         }
@@ -1070,22 +1111,23 @@ public record TestDataset(
         String dataFileName,
         String settingFileName,
         boolean allowSubFields,
-        @Nullable Map<String, String> typeMapping, // Override mappings read from mappings file
-        @Nullable Map<String, String> dynamicTypeMapping, // Define mappings not in the mapping files, but available from field-caps
+        @Nullable Map<String, String> typeMapping,
+        @Nullable Map<String, String> dynamicTypeMapping,
+        @Nullable String dynamic,
         List<String> inferenceEndpoints,
         List<EsqlCapabilities.Cap> requiredCapabilities
     ) {
 
         public TestDataset(String indexName) {
-            this(indexName, "mapping-" + indexName + ".json", indexName + ".csv", null, true, null, null, List.of(), List.of());
+            this(indexName, "mapping-" + indexName + ".json", indexName + ".csv", null, true, null, null, null, List.of(), List.of());
         }
 
         public TestDataset(String indexName, String mappingFileName, String dataFileName) {
-            this(indexName, mappingFileName, dataFileName, null, true, null, null, List.of(), List.of());
+            this(indexName, mappingFileName, dataFileName, null, true, null, null, null, List.of(), List.of());
         }
 
         public TestDataset(String indexName, String mappingFileName, String dataFileName, String settingFileName) {
-            this(indexName, mappingFileName, dataFileName, settingFileName, true, null, null, List.of(), List.of());
+            this(indexName, mappingFileName, dataFileName, settingFileName, true, null, null, null, List.of(), List.of());
         }
 
         public TestDataset withIndex(String indexName) {
@@ -1097,6 +1139,7 @@ public TestDataset withIndex(String indexName) {
                 allowSubFields,
                 typeMapping,
                 dynamicTypeMapping,
+                dynamic,
                 inferenceEndpoints,
                 requiredCapabilities
             );
@@ -1111,6 +1154,7 @@ public TestDataset withData(String dataFileName) {
                 allowSubFields,
                 typeMapping,
                 dynamicTypeMapping,
+                dynamic,
                 inferenceEndpoints,
                 requiredCapabilities
             );
@@ -1125,6 +1169,7 @@ public TestDataset noData() {
                 allowSubFields,
                 typeMapping,
                 dynamicTypeMapping,
+                dynamic,
                 inferenceEndpoints,
                 requiredCapabilities
             );
@@ -1139,6 +1184,7 @@ public TestDataset withSetting(String settingFileName) {
                 allowSubFields,
                 typeMapping,
                 dynamicTypeMapping,
+                dynamic,
                 inferenceEndpoints,
                 requiredCapabilities
             );
@@ -1153,11 +1199,20 @@ public TestDataset noSubfields() {
                 false,
                 typeMapping,
                 dynamicTypeMapping,
+                dynamic,
                 inferenceEndpoints,
                 requiredCapabilities
             );
         }
 
+        /**
+         * Overrides the types of fields in the mapping file. Each entry maps a field name to its new type
+         * (e.g. {@code Map.of("client_ip", "keyword")} changes client_ip from ip to keyword).
+         * A {@code null} value removes the field from the mapping entirely, making it unmapped for this index.
+         * <p>
+         * This affects both the Elasticsearch index mapping (via {@link CsvTestsDataLoader#readMappingFile}) and
+         * the in-memory mapping used by CSV unit tests.
+         */
         public TestDataset withTypeMapping(Map<String, String> typeMapping) {
             return new TestDataset(
                 indexName,
@@ -1167,11 +1222,18 @@ public TestDataset withTypeMapping(Map<String, String> typeMapping) {
                 allowSubFields,
                 typeMapping,
                 dynamicTypeMapping,
+                dynamic,
                 inferenceEndpoints,
                 requiredCapabilities
             );
         }
 
+        /**
+         * Adds field mappings that are not present in the mapping file, simulating fields discovered via field-caps
+         * at runtime (e.g. runtime fields). These fields are only added to the in-memory mapping used by CSV unit
+         * tests and are not included in the Elasticsearch index mapping. Fields already present in the mapping file
+         * are not overridden.
+         */
         public TestDataset withDynamicTypeMapping(Map<String, String> dynamicTypeMapping) {
             return new TestDataset(
                 indexName,
@@ -1181,6 +1243,26 @@ public TestDataset withDynamicTypeMapping(Map<String, String> dynamicTypeMapping
                 allowSubFields,
                 typeMapping,
                 dynamicTypeMapping,
+                dynamic,
+                inferenceEndpoints,
+                requiredCapabilities
+            );
+        }
+
+        /**
+         * Sets the "dynamic" mapping parameter (e.g. "false", "strict", "runtime").
+         * This prevents unmapped fields in the data from being automatically indexed.
+         */
+        public TestDataset withDynamic(String dynamic) {
+            return new TestDataset(
+                indexName,
+                mappingFileName,
+                dataFileName,
+                settingFileName,
+                allowSubFields,
+                typeMapping,
+                dynamicTypeMapping,
+                dynamic,
                 inferenceEndpoints,
                 requiredCapabilities
             );
@@ -1195,6 +1277,7 @@ public TestDataset withInferenceEndpoints(String... inferenceEndpoints) {
                 allowSubFields,
                 typeMapping,
                 dynamicTypeMapping,
+                dynamic,
                 List.of(inferenceEndpoints),
                 requiredCapabilities
             );
@@ -1209,6 +1292,7 @@ public TestDataset withRequiredCapabilities(EsqlCapabilities.Cap... requiredCapa
                 allowSubFields,
                 typeMapping,
                 dynamicTypeMapping,
+                dynamic,
                 inferenceEndpoints,
                 List.of(requiredCapabilities)
             );