### Elasticsearch Version

main

### Installed Plugins

_No response_

### Java Version

_bundled_

### OS Version

all

### Problem Description

Do this

### Steps to Reproduce

```
rm /tmp/evil
for a in {0..9}; do
   for b in {0..9}; do
      echo '{"index": {}}' >> /tmp/evil
      echo '{"a": '$a', "b": '$b'}' >> /tmp/evil
   done
done
echo >> /tmp/evil

curl -uelastic:password -XDELETE localhost:9200/test
curl -HContent-Type:application/json -uelastic:password -XPUT localhost:9200/test
curl -HContent-Type:application/json -uelastic:password -XPOST localhost:9200/test/_bulk --data-binary @/tmp/evil
curl -HContent-Type:application/json -uelastic:password -XPOST localhost:9200/test/_forcemerge?max_num_segments=1
curl -HContent-Type:application/json -uelastic:password -XPOST localhost:9200/test/_refresh

curl -HContent-Type:application/json -uelastic:password -XPOST localhost:9200/_esql?pretty -d'{
   "query": "FROM test | STATS MAX(a) BY a, b | KEEP a"
}'
```

### Logs (if relevant)

_No response_
x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/EnrichLookupService.java
listener = ActionListener.runBefore(listener, searchContext::close);
            SearchExecutionContext searchExecutionContext = searchContext.getSearchExecutionContext();
            MappedFieldType fieldType = searchExecutionContext.getFieldType(matchField);
            final SourceOperator queryOperator = switch (matchType) {
                case "match", "range" -> {
                    QueryList queryList = QueryList.termQueryList(fieldType, searchExecutionContext, inputBlock);
                    yield new EnrichQuerySourceOperator(queryList, searchExecutionContext.getIndexReader());
                }
                default -> throw new EsqlIllegalArgumentException("illegal match type " + matchType);
            };
            List<Operator> intermediateOperators = new ArrayList<>(extractFields.size() + 2);
            final ElementType[] mergingTypes = new ElementType[extractFields.size()];
            
            for (int i = 0; i < extractFields.size(); i++) {
                NamedExpression extractField = extractFields.get(i);
                final ElementType elementType = LocalExecutionPlanner.toElementType(extractField.dataType());
                mergingTypes[i] = elementType;
                var sources = ValueSources.sources(
                    List.of(searchContext),
                    extractField instanceof Alias a ? ((NamedExpression) a.child()).name() : extractField.name(),
                    EsqlDataTypes.isUnsupported(extractField.dataType()),
                    elementType
                );
                intermediateOperators.add(new ValuesSourceReaderOperator(sources, 0, extractField.name()));
            }
            
            intermediateOperators.add(droppingBlockOperator(extractFields.size() + 2, 0));
            boolean singleLeaf = searchContext.searcher().getLeafContexts().size() == 1;
            
            final int[] mergingChannels = IntStream.range(0, extractFields.size()).map(i -> i + 1).toArray();