ESQL: Check for errors while loading blocks (#129016)

Runs a sanity check after loading a block of values. Previously we were
doing a quick check if assertions were enabled. Now we do two quick
checks all the time. Better - we attach information about how a block
was loaded when there's a problem.

Relates to #128959

Author: nik9000

test/framework/src/main/java/org/elasticsearch/index/mapper/MapperServiceTestCase.java

@@ -103,7 +103,7 @@
 import static org.mockito.Mockito.mock;
 
 /**
- * Test case that lets you easilly build {@link MapperService} based on some
+ * Test case that lets you easily build {@link MapperService} based on some
  * mapping. Useful when you don't need to spin up an entire index but do
  * need most of the trapping of the mapping.
  */

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/lucene/ValuesSourceReaderOperator.java

@@ -27,7 +27,6 @@
 import org.elasticsearch.compute.operator.AbstractPageMappingOperator;
 import org.elasticsearch.compute.operator.DriverContext;
 import org.elasticsearch.compute.operator.Operator;
-import org.elasticsearch.core.Assertions;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Releasables;
 import org.elasticsearch.index.fieldvisitor.StoredFieldLoader;
@@ -161,12 +160,6 @@ public int get(int i) {
                     many.run();
                 }
             }
-            if (Assertions.ENABLED) {
-                for (int f = 0; f < fields.length; f++) {
-                    assert blocks[f].elementType() == ElementType.NULL || blocks[f].elementType() == fields[f].info.type
-                        : blocks[f].elementType() + " NOT IN (NULL, " + fields[f].info.type + ")";
-                }
-            }
             success = true;
             for (Block b : blocks) {
                 valuesLoaded += b.getTotalValueCount();
@@ -233,6 +226,7 @@ private void loadFromSingleLeaf(Block[] blocks, int shard, int segment, BlockLoa
                 BlockLoader.ColumnAtATimeReader columnAtATime = field.columnAtATime(ctx);
                 if (columnAtATime != null) {
                     blocks[f] = (Block) columnAtATime.read(loaderBlockFactory, docs);
+                    sanityCheckBlock(columnAtATime, docs.count(), blocks[f], f);
                 } else {
                     rowStrideReaders.add(
                         new RowStrideReaderWork(
@@ -282,6 +276,7 @@ private void loadFromSingleLeaf(Block[] blocks, int shard, int segment, BlockLoa
             }
             for (RowStrideReaderWork work : rowStrideReaders) {
                 blocks[work.offset] = work.build();
+                sanityCheckBlock(work.reader, docs.count(), blocks[work.offset], work.offset);
             }
         } finally {
             Releasables.close(rowStrideReaders);
@@ -385,6 +380,7 @@ void run() throws IOException {
                 try (Block targetBlock = fieldTypeBuilders[f].build()) {
                     target[f] = targetBlock.filter(backwards);
                 }
+                sanityCheckBlock(rowStride[f], docs.getPositionCount(), target[f], f);
             }
         }
 
@@ -561,6 +557,40 @@ protected Status status(long processNanos, int pagesProcessed, long rowsReceived
         return new Status(new TreeMap<>(readersBuilt), processNanos, pagesProcessed, rowsReceived, rowsEmitted, valuesLoaded);
     }
 
+    /**
+     * Quick checks for on the loaded block to make sure it looks reasonable.
+     * @param loader the object that did the loading - we use it to make error messages if the block is busted
+     * @param expectedPositions how many positions the block should have - it's as many as the incoming {@link Page} has
+     * @param block the block to sanity check
+     * @param field offset into the {@link #fields} array for the block being loaded
+     */
+    private void sanityCheckBlock(Object loader, int expectedPositions, Block block, int field) {
+        if (block.getPositionCount() != expectedPositions) {
+            throw new IllegalStateException(
+                sanityCheckBlockErrorPrefix(loader, block, field)
+                    + " has ["
+                    + block.getPositionCount()
+                    + "] positions instead of ["
+                    + expectedPositions
+                    + "]"
+            );
+        }
+        if (block.elementType() != ElementType.NULL && block.elementType() != fields[field].info.type) {
+            throw new IllegalStateException(
+                sanityCheckBlockErrorPrefix(loader, block, field)
+                    + "'s element_type ["
+                    + block.elementType()
+                    + "] NOT IN (NULL, "
+                    + fields[field].info.type
+                    + ")"
+            );
+        }
+    }
+
+    private String sanityCheckBlockErrorPrefix(Object loader, Block block, int field) {
+        return fields[field].info.name + "[" + loader + "]: " + block;
+    }
+
     public static class Status extends AbstractPageMappingOperator.Status {
         public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(
             Operator.Status.class,