ESQL: Refactor value reading so it can split Pages (#130573) (#130781)

This refactors our `ValuesSourceReaderOperator` so it can split pages
when it reads large values. It does not *actually* split the pages as
that's a bit tricky. But it sets the stage for the next PR that will
do so.

* Move `ValuesSourceReaderOperator` to it's own package
* Move many inner classes into their own top level classes
* Extend from `AbstractPageMappingToIteratorOperator` instead of
  `AbstractPageMappingToOperator`
  * This allows returning more than one `Page` per input `Page`
  * In this PR we still always return one `Page` per input `Page`
  * Make new `ReleasableIterator` subclasses to satisfy
    `AbstractPageMappingToIteratorOperator`
  * Change `status` of loading fields from `pages_processed` to
    `pages_received` and `pages_emitted`
* Fix a bug in `AbstractPageMappingToOperator` which can leak circuit
  breaker allocation if we fail to during `receive`. This isn't possible
  in the existing implementations but is possible
  in `ValuesSourceReaderOperator`.
* Add a test with large text fields. Right now it still comes back in
one page because we don't cut the pages.

Closes #130727

Also includes "Claim backported profile versions (#130187)"

Author: nik9000

benchmarks/src/main/java/org/elasticsearch/benchmark/compute/operator/ValuesSourceReaderBenchmark.java

@@ -41,7 +41,8 @@
 import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.lucene.LuceneSourceOperator;
 import org.elasticsearch.compute.lucene.ShardRefCounted;
-import org.elasticsearch.compute.lucene.ValuesSourceReaderOperator;
+import org.elasticsearch.compute.lucene.read.ValuesSourceReaderOperator;
+import org.elasticsearch.compute.lucene.read.ValuesSourceReaderOperatorStatus;
 import org.elasticsearch.compute.operator.topn.TopNOperator;
 import org.elasticsearch.core.IOUtils;
 import org.elasticsearch.index.IndexSettings;
@@ -343,7 +344,7 @@ public void benchmark() {
         );
         long sum = 0;
         for (Page page : pages) {
-            op.addInput(page);
+            op.addInput(page.shallowCopy());
             switch (name) {
                 case "long" -> {
                     LongVector values = op.getOutput().<LongBlock>getBlock(1).asVector();
@@ -411,7 +412,7 @@ public void benchmark() {
             throw new AssertionError("[" + layout + "][" + name + "] expected [" + expected + "] but was [" + sum + "]");
         }
         boolean foundStoredFieldLoader = false;
-        ValuesSourceReaderOperator.Status status = (ValuesSourceReaderOperator.Status) op.status();
+        ValuesSourceReaderOperatorStatus status = (ValuesSourceReaderOperatorStatus) op.status();
         for (Map.Entry<String, Integer> e : status.readersBuilt().entrySet()) {
             if (e.getKey().indexOf("stored_fields") >= 0) {
                 foundStoredFieldLoader = true;

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -209,6 +209,8 @@ static TransportVersion def(int id) {
     public static final TransportVersion ML_INFERENCE_ELASTIC_DENSE_TEXT_EMBEDDINGS_ADDED_8_19 = def(8_841_0_59);
     public static final TransportVersion ML_INFERENCE_COHERE_API_VERSION_8_19 = def(8_841_0_60);
     public static final TransportVersion ESQL_DOCUMENTS_FOUND_AND_VALUES_LOADED_8_19 = def(8_841_0_61);
+    public static final TransportVersion ESQL_PROFILE_INCLUDE_PLAN_8_19 = def(8_841_0_62);
+    public static final TransportVersion ESQL_SPLIT_ON_BIG_VALUES_8_19 = def(8_841_0_63);
     public static final TransportVersion V_9_0_0 = def(9_000_0_09);
     public static final TransportVersion INITIAL_ELASTICSEARCH_9_0_1 = def(9_000_0_10);
     public static final TransportVersion INITIAL_ELASTICSEARCH_9_0_2 = def(9_000_0_11);
@@ -325,6 +327,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion ML_INFERENCE_COHERE_API_VERSION = def(9_110_0_00);
     public static final TransportVersion ESQL_PROFILE_INCLUDE_PLAN = def(9_111_0_00);
     public static final TransportVersion MAPPINGS_IN_DATA_STREAMS = def(9_112_0_00);
+    public static final TransportVersion ESQL_SPLIT_ON_BIG_VALUES_9_1 = def(9_112_0_01);
 
     /*
      * STOP! READ THIS FIRST! No, really,

x-pack/plugin/esql/compute/src/main/java/module-info.java

@@ -36,4 +36,5 @@
     exports org.elasticsearch.compute.aggregation.table;
     exports org.elasticsearch.compute.data.sort;
     exports org.elasticsearch.compute.querydsl.query;
+    exports org.elasticsearch.compute.lucene.read;
 }