Use query source to detect non-null docs (elastic#133287)

One of the slowest parts of time-series queries is reading metric values 
- it accounts for 30% of the profiler time for the following query:

```
TS my*
| WHERE `metrics.system.memory.utilization` IS NOT NULL
        AND @timestamp >= "2025-07-25T14:55:59.000Z"
        AND @timestamp <= "2025-07-25T16:25:59.000Z"
| STATS AVG(AVG_OVER_TIME(`metrics.system.memory.utilization`)) BY host.name, BUCKET(@timestamp, 1h)
```

This is because the `metrics.system.memory.utilization` field is sparse, 
requiring iteration over its DISI to find value indices when reading
values. This change adds a flag named `nullsFiltered` to the column
reader, signaling that all target docs have values for the field. This
enables optimizations such as skipping value index lookups with DISI and
performing bulk copying.

We can safely do this because if the filter `WHERE 
metrics.system.memory.utilization IS NOT NULL` is pushed down to Lucene,
then every document returned from the Lucene operator will have a value
for the `metrics.system.memory.utilization` field.

Author: dnhatn

benchmarks/src/main/java/org/elasticsearch/benchmark/_nightly/esql/ValuesSourceReaderBenchmark.java

@@ -142,11 +142,26 @@ static void selfTest() {
     private static List<ValuesSourceReaderOperator.FieldInfo> fields(String name) {
         return switch (name) {
             case "3_stored_keywords" -> List.of(
-                new ValuesSourceReaderOperator.FieldInfo("keyword_1", ElementType.BYTES_REF, shardIdx -> blockLoader("stored_keyword_1")),
-                new ValuesSourceReaderOperator.FieldInfo("keyword_2", ElementType.BYTES_REF, shardIdx -> blockLoader("stored_keyword_2")),
-                new ValuesSourceReaderOperator.FieldInfo("keyword_3", ElementType.BYTES_REF, shardIdx -> blockLoader("stored_keyword_3"))
+                new ValuesSourceReaderOperator.FieldInfo(
+                    "keyword_1",
+                    ElementType.BYTES_REF,
+                    false,
+                    shardIdx -> blockLoader("stored_keyword_1")
+                ),
+                new ValuesSourceReaderOperator.FieldInfo(
+                    "keyword_2",
+                    ElementType.BYTES_REF,
+                    false,
+                    shardIdx -> blockLoader("stored_keyword_2")
+                ),
+                new ValuesSourceReaderOperator.FieldInfo(
+                    "keyword_3",
+                    ElementType.BYTES_REF,
+                    false,
+                    shardIdx -> blockLoader("stored_keyword_3")
+                )
             );
-            default -> List.of(new ValuesSourceReaderOperator.FieldInfo(name, elementType(name), shardIdx -> blockLoader(name)));
+            default -> List.of(new ValuesSourceReaderOperator.FieldInfo(name, elementType(name), false, shardIdx -> blockLoader(name)));
         };
     }
 

server/src/main/java/org/elasticsearch/index/mapper/AbstractShapeGeometryFieldMapper.java

@@ -98,7 +98,12 @@ protected void writeExtent(BlockLoader.IntBuilder builder, Extent extent) {
             public BlockLoader.AllReader reader(LeafReaderContext context) throws IOException {
                 return new BlockLoader.AllReader() {
                     @Override
-                    public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset) throws IOException {
+                    public BlockLoader.Block read(
+                        BlockLoader.BlockFactory factory,
+                        BlockLoader.Docs docs,
+                        int offset,
+                        boolean nullsFiltered
+                    ) throws IOException {
                         var binaryDocValues = context.reader().getBinaryDocValues(fieldName);
                         var reader = new GeometryDocValueReader();
                         try (var builder = factory.ints(docs.count() - offset)) {

server/src/main/java/org/elasticsearch/index/mapper/BlockDocValuesReader.java

@@ -130,7 +130,7 @@ static class SingletonLongs extends BlockDocValuesReader {
         }
 
         @Override
-        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
+        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFiltered) throws IOException {
             if (numericDocValues instanceof BlockLoader.OptionalColumnAtATimeReader direct) {
                 BlockLoader.Block result = direct.tryRead(factory, docs, offset);
                 if (result != null) {
@@ -179,7 +179,7 @@ static class Longs extends BlockDocValuesReader {
         }
 
         @Override
-        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
+        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFiltered) throws IOException {
             try (BlockLoader.LongBuilder builder = factory.longsFromDocValues(docs.count() - offset)) {
                 for (int i = offset; i < docs.count(); i++) {
                     int doc = docs.get(i);
@@ -260,7 +260,7 @@ private static class SingletonInts extends BlockDocValuesReader {
         }
 
         @Override
-        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
+        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFiltered) throws IOException {
             try (BlockLoader.IntBuilder builder = factory.intsFromDocValues(docs.count() - offset)) {
                 for (int i = offset; i < docs.count(); i++) {
                     int doc = docs.get(i);
@@ -303,7 +303,7 @@ private static class Ints extends BlockDocValuesReader {
         }
 
         @Override
-        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
+        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFiltered) throws IOException {
             try (BlockLoader.IntBuilder builder = factory.intsFromDocValues(docs.count() - offset)) {
                 for (int i = offset; i < docs.count(); i++) {
                     int doc = docs.get(i);
@@ -397,7 +397,7 @@ private static class SingletonDoubles extends BlockDocValuesReader {
         }
 
         @Override
-        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
+        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFiltered) throws IOException {
             try (BlockLoader.DoubleBuilder builder = factory.doublesFromDocValues(docs.count() - offset)) {
                 for (int i = offset; i < docs.count(); i++) {
                     int doc = docs.get(i);
@@ -442,7 +442,7 @@ private static class Doubles extends BlockDocValuesReader {
         }
 
         @Override
-        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
+        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFiltered) throws IOException {
             try (BlockLoader.DoubleBuilder builder = factory.doublesFromDocValues(docs.count() - offset)) {
                 for (int i = offset; i < docs.count(); i++) {
                     int doc = docs.get(i);
@@ -540,7 +540,7 @@ private abstract static class DenseVectorValuesBlockReader<T extends KnnVectorVa
         }
 
         @Override
-        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
+        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFiltered) throws IOException {
             // Doubles from doc values ensures that the values are in order
             try (BlockLoader.FloatBuilder builder = factory.denseVectors(docs.count() - offset, dimensions)) {
                 for (int i = offset; i < docs.count(); i++) {
@@ -710,7 +710,7 @@ private BlockLoader.Block readSingleDoc(BlockFactory factory, int docId) throws
         }
 
         @Override
-        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
+        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFiltered) throws IOException {
             if (docs.count() - offset == 1) {
                 return readSingleDoc(factory, docs.get(offset));
             }
@@ -761,7 +761,7 @@ private static class Ordinals extends BlockDocValuesReader {
         }
 
         @Override
-        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
+        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFiltered) throws IOException {
             if (docs.count() - offset == 1) {
                 return readSingleDoc(factory, docs.get(offset));
             }
@@ -875,7 +875,7 @@ private static class BytesRefsFromBinary extends BlockDocValuesReader {
         }
 
         @Override
-        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
+        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFiltered) throws IOException {
             try (BlockLoader.BytesRefBuilder builder = factory.bytesRefs(docs.count() - offset)) {
                 for (int i = offset; i < docs.count(); i++) {
                     int doc = docs.get(i);
@@ -988,7 +988,7 @@ public void read(int docId, BlockLoader.StoredFields storedFields, Builder build
         }
 
         @Override
-        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
+        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFiltered) throws IOException {
             try (BlockLoader.FloatBuilder builder = factory.denseVectors(docs.count() - offset, dimensions)) {
                 for (int i = offset; i < docs.count(); i++) {
                     int doc = docs.get(i);
@@ -1099,7 +1099,7 @@ private static class SingletonBooleans extends BlockDocValuesReader {
         }
 
         @Override
-        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
+        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFiltered) throws IOException {
             try (BlockLoader.BooleanBuilder builder = factory.booleansFromDocValues(docs.count() - offset)) {
                 int lastDoc = -1;
                 for (int i = offset; i < docs.count(); i++) {
@@ -1147,7 +1147,7 @@ private static class Booleans extends BlockDocValuesReader {
         }
 
         @Override
-        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
+        public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFiltered) throws IOException {
             try (BlockLoader.BooleanBuilder builder = factory.booleansFromDocValues(docs.count() - offset)) {
                 for (int i = offset; i < docs.count(); i++) {
                     int doc = docs.get(i);

server/src/main/java/org/elasticsearch/index/mapper/BlockLoader.java

@@ -43,8 +43,15 @@ interface Reader {
     interface ColumnAtATimeReader extends Reader {
         /**
          * Reads the values of all documents in {@code docs}.
-         */
-        BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException;
+         *
+         * @param nullsFiltered if {@code true}, then target docs are guaranteed to have a value for the field;
+         *                      otherwise, the guarantee is unknown. This enables optimizations for block loaders,
+         *                      treating the field as dense (every document has value) even if it is sparse in
+         *                      the index. For example, "FROM index | WHERE x != null | STATS sum(x)", after filtering out
+         *                      documents without value for field x, all target documents returned from the source operator
+         *                      will have a value for field x whether x is dense or sparse in the index.
+         */
+        BlockLoader.Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFiltered) throws IOException;
     }
 
     /**
@@ -166,7 +173,7 @@ public String toString() {
      */
     class ConstantNullsReader implements AllReader {
         @Override
-        public Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
+        public Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFiltered) throws IOException {
             return factory.constantNulls(docs.count() - offset);
         }
 
@@ -200,7 +207,7 @@ public Builder builder(BlockFactory factory, int expectedCount) {
             public ColumnAtATimeReader columnAtATimeReader(LeafReaderContext context) {
                 return new ColumnAtATimeReader() {
                     @Override
-                    public Block read(BlockFactory factory, Docs docs, int offset) {
+                    public Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFiltered) {
                         return factory.constantBytes(value, docs.count() - offset);
                     }
 
@@ -278,8 +285,8 @@ public ColumnAtATimeReader columnAtATimeReader(LeafReaderContext context) throws
             }
             return new ColumnAtATimeReader() {
                 @Override
-                public Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
-                    return reader.read(factory, docs, offset);
+                public Block read(BlockFactory factory, Docs docs, int offset, boolean nullsFiltered) throws IOException {
+                    return reader.read(factory, docs, offset, nullsFiltered);
                 }
 
                 @Override

server/src/main/java/org/elasticsearch/index/mapper/BooleanScriptBlockDocValuesReader.java

@@ -49,7 +49,8 @@ public int docId() {
     }
 
     @Override
-    public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset) throws IOException {
+    public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset, boolean nullsFiltered)
+        throws IOException {
         // Note that we don't emit falses before trues so we conform to the doc values contract and can use booleansFromDocValues
         try (BlockLoader.BooleanBuilder builder = factory.booleans(docs.count() - offset)) {
             for (int i = offset; i < docs.count(); i++) {

server/src/main/java/org/elasticsearch/index/mapper/DateScriptBlockDocValuesReader.java

@@ -49,7 +49,8 @@ public int docId() {
     }
 
     @Override
-    public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset) throws IOException {
+    public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset, boolean nullsFiltered)
+        throws IOException {
         // Note that we don't sort the values sort, so we can't use factory.longsFromDocValues
         try (BlockLoader.LongBuilder builder = factory.longs(docs.count() - offset)) {
             for (int i = offset; i < docs.count(); i++) {

server/src/main/java/org/elasticsearch/index/mapper/DoubleScriptBlockDocValuesReader.java

@@ -49,7 +49,8 @@ public int docId() {
     }
 
     @Override
-    public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset) throws IOException {
+    public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset, boolean nullsFiltered)
+        throws IOException {
         // Note that we don't sort the values sort, so we can't use factory.doublesFromDocValues
         try (BlockLoader.DoubleBuilder builder = factory.doubles(docs.count() - offset)) {
             for (int i = offset; i < docs.count(); i++) {

server/src/main/java/org/elasticsearch/index/mapper/IpScriptBlockDocValuesReader.java

@@ -49,7 +49,8 @@ public int docId() {
     }
 
     @Override
-    public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset) throws IOException {
+    public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset, boolean nullsFiltered)
+        throws IOException {
         // Note that we don't pre-sort our output so we can't use bytesRefsFromDocValues
         try (BlockLoader.BytesRefBuilder builder = factory.bytesRefs(docs.count() - offset)) {
             for (int i = offset; i < docs.count(); i++) {

server/src/main/java/org/elasticsearch/index/mapper/KeywordScriptBlockDocValuesReader.java

@@ -51,7 +51,8 @@ public int docId() {
     }
 
     @Override
-    public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset) throws IOException {
+    public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset, boolean nullsFiltered)
+        throws IOException {
         // Note that we don't pre-sort our output so we can't use bytesRefsFromDocValues
         try (BlockLoader.BytesRefBuilder builder = factory.bytesRefs(docs.count() - offset)) {
             for (int i = offset; i < docs.count(); i++) {

server/src/main/java/org/elasticsearch/index/mapper/LongScriptBlockDocValuesReader.java

@@ -49,7 +49,8 @@ public int docId() {
     }
 
     @Override
-    public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset) throws IOException {
+    public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset, boolean nullsFiltered)
+        throws IOException {
         // Note that we don't pre-sort our output so we can't use longsFromDocValues
         try (BlockLoader.LongBuilder builder = factory.longs(docs.count() - offset)) {
             for (int i = offset; i < docs.count(); i++) {