Use query source to detect non-null docs

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++) {

server/src/test/java/org/elasticsearch/index/mapper/AbstractShapeGeometryFieldMapperTests.java

@@ -125,7 +125,7 @@ private static void testBoundsBlockLoaderAux(
                     for (int j : array) {
                         expected.add(visitor.apply(geometries.get(j + currentIndex)).get());
                     }
-                    try (var block = (TestBlock) loader.reader(leaf).read(TestBlock.factory(), TestBlock.docs(array), 0)) {
+                    try (var block = (TestBlock) loader.reader(leaf).read(TestBlock.factory(), TestBlock.docs(array), 0, false)) {
                         for (int i = 0; i < block.size(); i++) {
                             intArrayResults.add(block.get(i));
                         }