Merge branch 'main' into svilen/126603

Author: svilen-mihaylov-elastic

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

@@ -25,6 +25,7 @@
 import org.apache.lucene.util.NumericUtils;
 import org.elasticsearch.common.breaker.NoopCircuitBreaker;
 import org.elasticsearch.common.lucene.Lucene;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.BytesRefBlock;
@@ -50,6 +51,7 @@
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.NumberFieldMapper;
 import org.elasticsearch.search.lookup.SearchLookup;
+import org.elasticsearch.xpack.esql.plugin.EsqlPlugin;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
 import org.openjdk.jmh.annotations.Fork;
@@ -335,7 +337,7 @@ public void benchmark() {
             fields(name),
             List.of(new ValuesSourceReaderOperator.ShardContext(reader, () -> {
                 throw new UnsupportedOperationException("can't load _source here");
-            })),
+            }, EsqlPlugin.STORED_FIELDS_SEQUENTIAL_PROPORTION.getDefault(Settings.EMPTY))),
             0
         );
         long sum = 0;

docs/changelog/127348.yaml

@@ -0,0 +1,5 @@
+pr: 127348
+summary: Speed loading stored fields
+area: ES|QL
+type: enhancement
+issues: []

docs/changelog/127527.yaml

@@ -0,0 +1,5 @@
+pr: 127527
+summary: "No, line noise isn't a valid ip"
+area: ES|QL
+type: bug
+issues: []

docs/reference/elasticsearch/index-settings/index-modules.md

@@ -49,7 +49,7 @@ $$$index-codec$$$ `index.codec`
 
 $$$index-mode-setting$$$ `index.mode`
 :   The `index.mode` setting is used to control settings applied in specific domains like ingestion of time series data or logs. Different mutually exclusive modes exist, which are used to apply settings or default values controlling indexing of documents, sorting and other parameters whose value affects indexing or query performance.
-        
+
         **Example**
 
       ```console
@@ -248,3 +248,8 @@ $$$index-final-pipeline$$$
 
 $$$index-hidden$$$ `index.hidden`
 :   Indicates whether the index should be hidden by default. Hidden indices are not returned by default when using a wildcard expression. This behavior is controlled per request through the use of the `expand_wildcards` parameter. Possible values are `true` and `false` (default).
+
+$$$index-esql-stored-fields-sequential-proportion$$$
+
+`index.esql.stored_fields_sequential_proportion`
+:   Tuning parameter for deciding when {{esql}} will load [Stored fields](/reference/elasticsearch/rest-apis/retrieve-selected-fields.md#stored-fields) using a strategy tuned for loading dense sequence of documents. Allows values between 0.0 and 1.0 and defaults to 0.2. Indices with documents smaller than 10kb may see speed improvements loading `text` fields by setting this lower.

muted-tests.yml

@@ -444,6 +444,15 @@ tests:
 - class: org.elasticsearch.action.admin.cluster.state.TransportClusterStateActionDisruptionIT
   method: testLocalRequestWaitsForMetadata
   issue: https://github.com/elastic/elasticsearch/issues/127466
+- class: org.elasticsearch.xpack.esql.type.EsqlDataTypeConverterTests
+  method: testSuggestedCast
+  issue: https://github.com/elastic/elasticsearch/issues/127535
+- class: org.elasticsearch.xpack.esql.qa.single_node.GenerativeIT
+  method: test
+  issue: https://github.com/elastic/elasticsearch/issues/127536
+- class: org.elasticsearch.xpack.esql.qa.mixed.MixedClusterEsqlSpecIT
+  method: test {union_types.MultiIndexSortIpStringEval ASYNC}
+  issue: https://github.com/elastic/elasticsearch/issues/127537
 
 # Examples:
 #

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/aggregation/blockhash/TimeSeriesBlockHash.java

@@ -30,6 +30,8 @@
 import org.elasticsearch.core.ReleasableIterator;
 import org.elasticsearch.core.Releasables;
 
+import java.util.Objects;
+
 /**
  * An optimized block hash that receives two blocks: tsid and timestamp, which are sorted.
  * Since the incoming data is sorted, this block hash appends the incoming data to the internal arrays without lookup.
@@ -39,7 +41,7 @@ public final class TimeSeriesBlockHash extends BlockHash {
     private final int tsHashChannel;
     private final int timestampIntervalChannel;
 
-    private int lastTsidPosition = 0;
+    private final BytesRef lastTsid = new BytesRef();
     private final BytesRefArrayWithSize tsidArray;
 
     private long lastTimestamp;
@@ -62,77 +64,44 @@ public void close() {
         Releasables.close(tsidArray, timestampArray, perTsidCountArray);
     }
 
-    private OrdinalBytesRefVector getTsidVector(Page page) {
-        BytesRefBlock block = page.getBlock(tsHashChannel);
-        var ordinalBlock = block.asOrdinals();
-        if (ordinalBlock == null) {
-            throw new IllegalStateException("expected ordinal block for tsid");
-        }
-        var ordinalVector = ordinalBlock.asVector();
-        if (ordinalVector == null) {
-            throw new IllegalStateException("expected ordinal vector for tsid");
-        }
-        return ordinalVector;
-    }
-
-    private LongVector getTimestampVector(Page page) {
-        final LongBlock timestampsBlock = page.getBlock(timestampIntervalChannel);
-        LongVector timestampsVector = timestampsBlock.asVector();
-        if (timestampsVector == null) {
-            throw new IllegalStateException("expected long vector for timestamp");
-        }
-        return timestampsVector;
-    }
-
     @Override
     public void add(Page page, GroupingAggregatorFunction.AddInput addInput) {
-        final BytesRefVector tsidDict;
-        final IntVector tsidOrdinals;
-        {
-            final var tsidVector = getTsidVector(page);
-            tsidDict = tsidVector.getDictionaryVector();
-            tsidOrdinals = tsidVector.getOrdinalsVector();
-        }
-        try (var ordsBuilder = blockFactory.newIntVectorBuilder(tsidOrdinals.getPositionCount())) {
+        final BytesRefBlock tsidBlock = page.getBlock(tsHashChannel);
+        final BytesRefVector tsidVector = Objects.requireNonNull(tsidBlock.asVector(), "tsid input must be a vector");
+        final LongBlock timestampBlock = page.getBlock(timestampIntervalChannel);
+        final LongVector timestampVector = Objects.requireNonNull(timestampBlock.asVector(), "timestamp input must be a vector");
+        try (var ordsBuilder = blockFactory.newIntVectorBuilder(tsidVector.getPositionCount())) {
             final BytesRef spare = new BytesRef();
-            final BytesRef lastTsid = new BytesRef();
-            final LongVector timestampVector = getTimestampVector(page);
-            int lastOrd = -1;
-            for (int i = 0; i < tsidOrdinals.getPositionCount(); i++) {
-                final int newOrd = tsidOrdinals.getInt(i);
-                boolean newGroup = false;
-                if (lastOrd != newOrd) {
-                    final var newTsid = tsidDict.getBytesRef(newOrd, spare);
-                    if (positionCount() == 0) {
-                        newGroup = true;
-                    } else if (lastOrd == -1) {
-                        tsidArray.get(lastTsidPosition, lastTsid);
-                        newGroup = lastTsid.equals(newTsid) == false;
-                    } else {
-                        newGroup = true;
-                    }
-                    if (newGroup) {
-                        endTsidGroup();
-                        lastTsidPosition = tsidArray.count;
-                        tsidArray.append(newTsid);
-                    }
-                    lastOrd = newOrd;
-                }
+            // TODO: optimize incoming ordinal block
+            for (int i = 0; i < tsidVector.getPositionCount(); i++) {
+                final BytesRef tsid = tsidVector.getBytesRef(i, spare);
                 final long timestamp = timestampVector.getLong(i);
-                if (newGroup || timestamp != lastTimestamp) {
-                    assert newGroup || lastTimestamp >= timestamp : "@timestamp goes backward " + lastTimestamp + " < " + timestamp;
-                    timestampArray.append(timestamp);
-                    lastTimestamp = timestamp;
-                    currentTimestampCount++;
-                }
-                ordsBuilder.appendInt(timestampArray.count - 1);
+                ordsBuilder.appendInt(addOnePosition(tsid, timestamp));
             }
             try (var ords = ordsBuilder.build()) {
                 addInput.add(0, ords);
             }
         }
     }
 
+    private int addOnePosition(BytesRef tsid, long timestamp) {
+        boolean newGroup = false;
+        if (positionCount() == 0 || lastTsid.equals(tsid) == false) {
+            assert positionCount() == 0 || lastTsid.compareTo(tsid) < 0 : "tsid goes backward ";
+            endTsidGroup();
+            tsidArray.append(tsid);
+            tsidArray.get(tsidArray.count - 1, lastTsid);
+            newGroup = true;
+        }
+        if (newGroup || timestamp != lastTimestamp) {
+            assert newGroup || lastTimestamp >= timestamp : "@timestamp goes backward " + lastTimestamp + " < " + timestamp;
+            timestampArray.append(timestamp);
+            lastTimestamp = timestamp;
+            currentTimestampCount++;
+        }
+        return positionCount() - 1;
+    }
+
     private void endTsidGroup() {
         if (currentTimestampCount > 0) {
             perTsidCountArray.append(currentTimestampCount);

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/data/OrdinalBytesRefBlock.java

@@ -75,7 +75,7 @@ public BytesRef getBytesRef(int valueIndex, BytesRef dest) {
     }
 
     @Override
-    public OrdinalBytesRefVector asVector() {
+    public BytesRefVector asVector() {
         IntVector vector = ordinals.asVector();
         if (vector != null) {
             return new OrdinalBytesRefVector(vector, bytes);

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

@@ -107,7 +107,7 @@ public String describe() {
      */
     public record FieldInfo(String name, ElementType type, IntFunction<BlockLoader> blockLoader) {}
 
-    public record ShardContext(IndexReader reader, Supplier<SourceLoader> newSourceLoader) {}
+    public record ShardContext(IndexReader reader, Supplier<SourceLoader> newSourceLoader, double storedFieldsSequentialProportion) {}
 
     private final FieldWork[] fields;
     private final List<ShardContext> shardContexts;
@@ -247,8 +247,9 @@ private void loadFromSingleLeaf(Block[] blocks, int shard, int segment, BlockLoa
             }
 
             SourceLoader sourceLoader = null;
+            ShardContext shardContext = shardContexts.get(shard);
             if (storedFieldsSpec.requiresSource()) {
-                sourceLoader = shardContexts.get(shard).newSourceLoader.get();
+                sourceLoader = shardContext.newSourceLoader.get();
                 storedFieldsSpec = storedFieldsSpec.merge(new StoredFieldsSpec(true, false, sourceLoader.requiredStoredFields()));
             }
 
@@ -261,7 +262,7 @@ private void loadFromSingleLeaf(Block[] blocks, int shard, int segment, BlockLoa
                 );
             }
             StoredFieldLoader storedFieldLoader;
-            if (useSequentialStoredFieldsReader(docs)) {
+            if (useSequentialStoredFieldsReader(docs, shardContext.storedFieldsSequentialProportion())) {
                 storedFieldLoader = StoredFieldLoader.fromSpecSequential(storedFieldsSpec);
                 trackStoredFields(storedFieldsSpec, true);
             } else {
@@ -438,9 +439,13 @@ public void close() {
      * Is it more efficient to use a sequential stored field reader
      * when reading stored fields for the documents contained in {@code docIds}?
      */
-    private boolean useSequentialStoredFieldsReader(BlockLoader.Docs docs) {
+    private boolean useSequentialStoredFieldsReader(BlockLoader.Docs docs, double storedFieldsSequentialProportion) {
         int count = docs.count();
-        return count >= SEQUENTIAL_BOUNDARY && docs.get(count - 1) - docs.get(0) == count - 1;
+        if (count < SEQUENTIAL_BOUNDARY) {
+            return false;
+        }
+        int range = docs.get(count - 1) - docs.get(0);
+        return range * storedFieldsSequentialProportion <= count;
     }
 
     private void trackStoredFields(StoredFieldsSpec spec, boolean sequential) {

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/TimeSeriesAggregationOperator.java

@@ -14,7 +14,6 @@
 import org.elasticsearch.compute.aggregation.GroupingAggregatorEvaluationContext;
 import org.elasticsearch.compute.aggregation.TimeSeriesGroupingAggregatorEvaluationContext;
 import org.elasticsearch.compute.aggregation.blockhash.BlockHash;
-import org.elasticsearch.compute.aggregation.blockhash.TimeSeriesBlockHash;
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.ElementType;
 import org.elasticsearch.compute.data.LongBlock;
@@ -31,7 +30,6 @@ public class TimeSeriesAggregationOperator extends HashAggregationOperator {
 
     public record Factory(
         Rounding.Prepared timeBucket,
-        boolean sortedInput,
         List<BlockHash.GroupSpec> groups,
         AggregatorMode aggregatorMode,
         List<GroupingAggregator.Factory> aggregators,
@@ -40,18 +38,17 @@ public record Factory(
         @Override
         public Operator get(DriverContext driverContext) {
             // TODO: use TimeSeriesBlockHash when possible
-            return new TimeSeriesAggregationOperator(timeBucket, aggregators, () -> {
-                if (sortedInput && groups.size() == 2) {
-                    return new TimeSeriesBlockHash(groups.get(0).channel(), groups.get(1).channel(), driverContext.blockFactory());
-                } else {
-                    return BlockHash.build(
-                        groups,
-                        driverContext.blockFactory(),
-                        maxPageSize,
-                        true // we can enable optimizations as the inputs are vectors
-                    );
-                }
-            }, driverContext);
+            return new TimeSeriesAggregationOperator(
+                timeBucket,
+                aggregators,
+                () -> BlockHash.build(
+                    groups,
+                    driverContext.blockFactory(),
+                    maxPageSize,
+                    true // we can enable optimizations as the inputs are vectors
+                ),
+                driverContext
+            );
         }
 
         @Override

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/OperatorTests.java

@@ -198,7 +198,7 @@ public String toString() {
                 operators.add(
                     new OrdinalsGroupingOperator(
                         shardIdx -> new KeywordFieldMapper.KeywordFieldType("g").blockLoader(mockBlContext()),
-                        List.of(new ValuesSourceReaderOperator.ShardContext(reader, () -> SourceLoader.FROM_STORED_SOURCE)),
+                        List.of(new ValuesSourceReaderOperator.ShardContext(reader, () -> SourceLoader.FROM_STORED_SOURCE, 0.2)),
                         ElementType.BYTES_REF,
                         0,
                         gField,