ESQL: Check for errors while loading blocks

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,24 @@ 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(loader + ": " + block + " didn't have [" + expectedPositions + "] positions");
+ }
+ if (block.elementType() != ElementType.NULL && block.elementType() != fields[field].info.type) {
+ throw new IllegalStateException(
+ loader + ": " + block + "'s element_type [" + block.elementType() + "] NOT IN (NULL, " + fields[field].info.type + ")"
+ );
+ }
+ }
+
  public static class Status extends AbstractPageMappingOperator.Status {
  public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(
  Operator.Status.class,