ESQL: Fix loading constant_keyword when unset (elastic#101496)

When a `constant_keyword` field is first created you don't *have* to set
it's value - it'll default to `null` and then take on the value of the
first document that contains the field. Our block loading code didn't
like this state and would fail to run the query. This fixes that.

Closes elastic#101455

x-pack/plugin/esql/qa/server/single-node/src/yamlRestTest/resources/rest-api-spec/test/30_types.yml

@@ -53,6 +53,39 @@ constant_keyword:
   - length: {values: 1}
   - match: {values.0.0: 17}
 
+---
+constant_keyword with null value:
+  - do:
+      indices.create:
+        index:  test
+        body:
+          mappings:
+            properties:
+              kind:
+                type: constant_keyword
+              color:
+                type: keyword
+
+  - do:
+      bulk:
+        index: test
+        refresh: true
+        body:
+          - { "index": { } }
+          - { "color": "red" }
+
+  - do:
+      esql.query:
+        body:
+          query: 'from test | limit 1'
+  - match: { columns.0.name: color }
+  - match: { columns.0.type: keyword }
+  - match: { columns.1.name: kind }
+  - match: { columns.1.type: keyword }
+  - length: { values: 1 }
+  - match: { values.0.0: red }
+  - match: { values.0.1: null }
+
 ---
 multivalued keyword:
   - do:

x-pack/plugin/mapper-constant-keyword/src/main/java/org/elasticsearch/xpack/constantkeyword/mapper/ConstantKeywordFieldMapper.java

@@ -138,6 +138,9 @@ public String familyTypeName() {
         @Override
         public BlockLoader blockLoader(BlockLoaderContext blContext) {
             // TODO build a constant block directly
+            if (value == null) {
+                return BlockDocValuesReader.nulls();
+            }
             BytesRef bytes = new BytesRef(value);
             return context -> new BlockDocValuesReader() {
                 private int docId;

x-pack/plugin/mapper-constant-keyword/src/test/java/org/elasticsearch/xpack/constantkeyword/mapper/ConstantKeywordFieldMapperTests.java

@@ -7,12 +7,18 @@
 
 package org.elasticsearch.xpack.constantkeyword.mapper;
 
+import org.apache.lucene.index.DirectoryReader;
 import org.apache.lucene.search.MatchNoDocsQuery;
 import org.apache.lucene.search.Query;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.tests.index.RandomIndexWriter;
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.compress.CompressedXContent;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.CheckedFunction;
+import org.elasticsearch.index.mapper.BlockDocValuesReader;
+import org.elasticsearch.index.mapper.BlockLoader;
 import org.elasticsearch.index.mapper.DocumentMapper;
 import org.elasticsearch.index.mapper.DocumentParsingException;
 import org.elasticsearch.index.mapper.LuceneDocument;
@@ -22,7 +28,9 @@
 import org.elasticsearch.index.mapper.MapperService.MergeReason;
 import org.elasticsearch.index.mapper.MapperTestCase;
 import org.elasticsearch.index.mapper.ParsedDocument;
+import org.elasticsearch.index.mapper.TestBlock;
 import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.search.lookup.SearchLookup;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xpack.constantkeyword.ConstantKeywordMapperPlugin;
 import org.elasticsearch.xpack.constantkeyword.mapper.ConstantKeywordFieldMapper.ConstantKeywordFieldType;
@@ -31,11 +39,13 @@
 import java.io.IOException;
 import java.util.Collection;
 import java.util.List;
+import java.util.Set;
 import java.util.function.Function;
 
 import static org.elasticsearch.index.mapper.MapperService.INDEX_MAPPING_TOTAL_FIELDS_LIMIT_SETTING;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.nullValue;
 
 public class ConstantKeywordFieldMapperTests extends MapperTestCase {
 
@@ -214,6 +224,62 @@ protected boolean allowsNullValues() {
         return false;   // null is an error for constant keyword
     }
 
+    /**
+     * Test loading blocks when there is no defined value. This is allowed
+     * for newly created indices that haven't received any documents that
+     * contain the field.
+     */
+    public void testNullValueBlockLoaderReadValues() throws IOException {
+        testNullBlockLoader(blockReader -> (TestBlock) blockReader.readValues(TestBlock.FACTORY, TestBlock.docs(0)));
+    }
+
+    /**
+     * Test loading blocks when there is no defined value. This is allowed
+     * for newly created indices that haven't received any documents that
+     * contain the field.
+     */
+    public void testNullValueBlockLoaderReadValuesFromSingleDoc() throws IOException {
+        testNullBlockLoader(blockReader -> {
+            TestBlock block = (TestBlock) blockReader.builder(TestBlock.FACTORY, 1);
+            blockReader.readValuesFromSingleDoc(0, block);
+            return block;
+        });
+    }
+
+    private void testNullBlockLoader(CheckedFunction<BlockDocValuesReader, TestBlock, IOException> body) throws IOException {
+        MapperService mapper = createMapperService(syntheticSourceMapping(b -> {
+            b.startObject("field");
+            b.field("type", "constant_keyword");
+            b.endObject();
+        }));
+        BlockLoader loader = mapper.fieldType("field").blockLoader(new MappedFieldType.BlockLoaderContext() {
+            @Override
+            public String indexName() {
+                throw new UnsupportedOperationException();
+            }
+
+            @Override
+            public SearchLookup lookup() {
+                throw new UnsupportedOperationException();
+            }
+
+            @Override
+            public Set<String> sourcePaths(String name) {
+                return mapper.mappingLookup().sourcePaths(name);
+            }
+        });
+        try (Directory directory = newDirectory()) {
+            RandomIndexWriter iw = new RandomIndexWriter(random(), directory);
+            LuceneDocument doc = mapper.documentMapper().parse(source(b -> {})).rootDoc();
+            iw.addDocument(doc);
+            iw.close();
+            try (DirectoryReader reader = DirectoryReader.open(directory)) {
+                TestBlock block = body.apply(loader.reader(reader.leaves().get(0)));
+                assertThat(block.get(0), nullValue());
+            }
+        }
+    }
+
     @Override
     protected SyntheticSourceSupport syntheticSourceSupport(boolean ignoreMalformed) {
         assertFalse("constant_keyword doesn't support ignore_malformed", ignoreMalformed);