Merge branch 'main' into 2025/09/09/cluster-applier-thread-watchdog

Author: DaveCTurner

docs/changelog/133861.yaml

@@ -0,0 +1,5 @@
+pr: 133861
+summary: Implementing latency improvements for EIS integration
+area: Machine Learning
+type: bug
+issues: []

libs/exponential-histogram/build.gradle

@@ -10,6 +10,7 @@
 // TODO: publish this when ready?
 //apply plugin: 'elasticsearch.publish'
 apply plugin: 'elasticsearch.build'
+apply plugin: 'elasticsearch.internal-test-artifact'
 
 dependencies {
   api project(':libs:core')

libs/exponential-histogram/src/test/java/org/elasticsearch/exponentialhistogram/ExponentialHistogramTestCase.java

@@ -70,7 +70,7 @@ ExponentialHistogram createAutoReleasedHistogram(Consumer<ExponentialHistogramBu
         return result;
     }
 
-    ExponentialHistogram createAutoReleasedHistogram(int numBuckets, double... values) {
+    protected ExponentialHistogram createAutoReleasedHistogram(int numBuckets, double... values) {
         ReleasableExponentialHistogram result = ExponentialHistogram.create(numBuckets, breaker(), values);
         releaseBeforeEnd.add(result);
         return result;

libs/exponential-histogram/src/test/java/org/elasticsearch/exponentialhistogram/QuantileAccuracyTests.java

@@ -221,7 +221,7 @@ private void testDistributionQuantileAccuracy(RealDistribution distribution) {
         testQuantileAccuracy(values, bucketCount);
     }
 
-    private static double[] generateSamples(RealDistribution distribution, int sampleSize) {
+    public static double[] generateSamples(RealDistribution distribution, int sampleSize) {
         double[] values = new double[sampleSize];
         for (int i = 0; i < sampleSize; i++) {
             values[i] = distribution.sample();
@@ -276,7 +276,7 @@ private double testQuantileAccuracy(double[] values, int bucketCount) {
      * The error depends on the raw values put into the histogram and the number of buckets allowed.
      * This is an implementation of the error bound computation proven by Theorem 3 in the <a href="https://arxiv.org/pdf/2004.08604">UDDSketch paper</a>
      */
-    private static double getMaximumRelativeError(double[] values, int bucketCount) {
+    public static double getMaximumRelativeError(double[] values, int bucketCount) {
         HashSet<Long> usedPositiveIndices = new HashSet<>();
         HashSet<Long> usedNegativeIndices = new HashSet<>();
         int bestPossibleScale = MAX_SCALE;

muted-tests.yml

@@ -474,9 +474,6 @@ tests:
 - class: org.elasticsearch.xpack.esql.qa.multi_node.EsqlClientYamlIT
   method: test {p0=esql/60_usage/Basic ESQL usage output (telemetry) non-snapshot version}
   issue: https://github.com/elastic/elasticsearch/issues/133461
-- class: org.elasticsearch.xpack.esql.action.LookupJoinTypesIT
-  method: testLookupJoinOthers
-  issue: https://github.com/elastic/elasticsearch/issues/133480
 - class: org.elasticsearch.xpack.esql.action.CrossClusterAsyncQueryStopIT
   method: testStopQueryLocal
   issue: https://github.com/elastic/elasticsearch/issues/133481

qa/vector/src/main/java/org/elasticsearch/test/knn/KnnIndexer.java

@@ -32,9 +32,18 @@
 import org.apache.lucene.index.MergePolicy;
 import org.apache.lucene.index.VectorEncoding;
 import org.apache.lucene.index.VectorSimilarityFunction;
+import org.apache.lucene.store.Directory;
 import org.apache.lucene.store.FSDirectory;
+import org.apache.lucene.store.IOContext;
+import org.apache.lucene.store.IndexInput;
+import org.apache.lucene.store.MMapDirectory;
+import org.apache.lucene.store.NIOFSDirectory;
+import org.apache.lucene.store.NativeFSLockFactory;
 import org.apache.lucene.util.PrintStreamInfoStream;
 import org.elasticsearch.common.io.Channels;
+import org.elasticsearch.core.IOUtils;
+import org.elasticsearch.index.store.LuceneFilesExtensions;
+import org.elasticsearch.index.store.Store;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
@@ -124,7 +133,7 @@ public boolean isEnabled(String component) {
 
         long start = System.nanoTime();
         AtomicInteger numDocsIndexed = new AtomicInteger();
-        try (FSDirectory dir = FSDirectory.open(indexPath); IndexWriter iw = new IndexWriter(dir, iwc);) {
+        try (Directory dir = getDirectory(indexPath); IndexWriter iw = new IndexWriter(dir, iwc)) {
             for (Path docsPath : this.docsPath) {
                 int dim = this.dim;
                 try (FileChannel in = FileChannel.open(docsPath)) {
@@ -212,7 +221,7 @@ public boolean isEnabled(String component) {
         iwc.setCodec(codec);
         logger.debug("KnnIndexer: forceMerge in {}", indexPath);
         long startNS = System.nanoTime();
-        try (IndexWriter iw = new IndexWriter(FSDirectory.open(indexPath), iwc)) {
+        try (IndexWriter iw = new IndexWriter(getDirectory(indexPath), iwc)) {
             iw.forceMerge(1);
         }
         long endNS = System.nanoTime();
@@ -221,6 +230,14 @@ public boolean isEnabled(String component) {
         results.forceMergeTimeMS = TimeUnit.NANOSECONDS.toMillis(elapsedNSec);
     }
 
+    static Directory getDirectory(Path indexPath) throws IOException {
+        Directory dir = FSDirectory.open(indexPath);
+        if (dir instanceof MMapDirectory mmapDir) {
+            return new HybridDirectory(mmapDir);
+        }
+        return dir;
+    }
+
     static class IndexerThread extends Thread {
         private final IndexWriter iw;
         private final AtomicInteger numDocsIndexed;
@@ -358,4 +375,64 @@ synchronized void next(byte[] dest) throws IOException {
             bytes.get(dest);
         }
     }
+
+    // Copy of Elastic's HybridDirectory which extends NIOFSDirectory and uses MMapDirectory for certain files.
+    static final class HybridDirectory extends NIOFSDirectory {
+        private final MMapDirectory delegate;
+
+        HybridDirectory(MMapDirectory delegate) throws IOException {
+            super(delegate.getDirectory(), NativeFSLockFactory.INSTANCE);
+            this.delegate = delegate;
+        }
+
+        @Override
+        public IndexInput openInput(String name, IOContext context) throws IOException {
+            if (useDelegate(name, context)) {
+                // we need to do these checks on the outer directory since the inner doesn't know about pending deletes
+                ensureOpen();
+                ensureCanRead(name);
+                // we switch the context here since mmap checks for the READONCE context by identity
+                context = context == Store.READONCE_CHECKSUM ? IOContext.READONCE : context;
+                // we only use the mmap to open inputs. Everything else is managed by the NIOFSDirectory otherwise
+                // we might run into trouble with files that are pendingDelete in one directory but still
+                // listed in listAll() from the other. We on the other hand don't want to list files from both dirs
+                // and intersect for perf reasons.
+                return delegate.openInput(name, context);
+            } else {
+                return super.openInput(name, context);
+            }
+        }
+
+        @Override
+        public void close() throws IOException {
+            IOUtils.close(super::close, delegate);
+        }
+
+        private static String getExtension(String name) {
+            // Unlike FileSwitchDirectory#getExtension, we treat `tmp` as a normal file extension, which can have its own rules for mmaping.
+            final int lastDotIndex = name.lastIndexOf('.');
+            if (lastDotIndex == -1) {
+                return "";
+            } else {
+                return name.substring(lastDotIndex + 1);
+            }
+        }
+
+        static boolean useDelegate(String name, IOContext ioContext) {
+            if (ioContext == Store.READONCE_CHECKSUM) {
+                // If we're just reading the footer for the checksum then mmap() isn't really necessary, and it's desperately inefficient
+                // if pre-loading is enabled on this file.
+                return false;
+            }
+
+            final LuceneFilesExtensions extension = LuceneFilesExtensions.fromExtension(getExtension(name));
+            if (extension == null || extension.shouldMmap() == false) {
+                // Other files are either less performance-sensitive (e.g. stored field index, norms metadata)
+                // or are large and have a random access pattern and mmap leads to page cache trashing
+                // (e.g. stored fields and term vectors).
+                return false;
+            }
+            return true;
+        }
+    }
 }

qa/vector/src/main/java/org/elasticsearch/test/knn/KnnSearcher.java

@@ -52,7 +52,6 @@
 import org.apache.lucene.search.Weight;
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.store.FSDirectory;
-import org.apache.lucene.store.MMapDirectory;
 import org.apache.lucene.util.BitSet;
 import org.apache.lucene.util.BitSetIterator;
 import org.apache.lucene.util.FixedBitSet;
@@ -178,7 +177,7 @@ void runSearch(KnnIndexTester.Results finalResults, boolean earlyTermination) th
             );
             KnnIndexer.VectorReader targetReader = KnnIndexer.VectorReader.create(input, dim, vectorEncoding, offsetByteSize);
             long startNS;
-            try (MMapDirectory dir = new MMapDirectory(indexPath)) {
+            try (Directory dir = KnnIndexer.getDirectory(indexPath)) {
                 try (DirectoryReader reader = DirectoryReader.open(dir)) {
                     IndexSearcher searcher = searchThreads > 1 ? new IndexSearcher(reader, executorService) : new IndexSearcher(reader);
                     byte[] targetBytes = new byte[dim];

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

@@ -343,15 +343,8 @@ static TransportVersion def(int id) {
     public static final TransportVersion ML_INFERENCE_LLAMA_ADDED = def(9_125_0_00);
     public static final TransportVersion SHARD_WRITE_LOAD_IN_CLUSTER_INFO = def(9_126_0_00);
     public static final TransportVersion ESQL_SAMPLE_OPERATOR_STATUS = def(9_127_0_00);
-    public static final TransportVersion PROJECT_RESERVED_STATE_MOVE_TO_REGISTRY = def(9_147_0_00);
-    public static final TransportVersion STREAMS_ENDPOINT_PARAM_RESTRICTIONS = def(9_148_0_00);
-    public static final TransportVersion RESOLVE_INDEX_MODE_FILTER = def(9_149_0_00);
-    public static final TransportVersion SEMANTIC_QUERY_MULTIPLE_INFERENCE_IDS = def(9_150_0_00);
-    public static final TransportVersion ESQL_LOOKUP_JOIN_PRE_JOIN_FILTER = def(9_151_0_00);
-    public static final TransportVersion INFERENCE_API_DISABLE_EIS_RATE_LIMITING = def(9_152_0_00);
-    public static final TransportVersion GEMINI_THINKING_BUDGET_ADDED = def(9_153_0_00);
-    public static final TransportVersion VISIT_PERCENTAGE = def(9_154_0_00);
     public static final TransportVersion TIME_SERIES_TELEMETRY = def(9_155_0_00);
+    public static final TransportVersion INFERENCE_API_EIS_DIAGNOSTICS = def(9_156_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/action/admin/indices/resolve/ResolveIndexAction.java

@@ -10,7 +10,6 @@
 package org.elasticsearch.action.admin.indices.resolve;
 
 import org.elasticsearch.TransportVersion;
-import org.elasticsearch.TransportVersions;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.ActionResponse;
@@ -78,6 +77,7 @@ public class ResolveIndexAction extends ActionType<ResolveIndexAction.Response>
     public static final RemoteClusterActionType<Response> REMOTE_TYPE = new RemoteClusterActionType<>(NAME, Response::new);
 
     private static final TransportVersion RESOLVE_INDEX_MODE_ADDED = TransportVersion.fromName("resolve_index_mode_added");
+    private static final TransportVersion RESOLVE_INDEX_MODE_FILTER = TransportVersion.fromName("resolve_index_mode_filter");
 
     private ResolveIndexAction() {
         super(NAME);
@@ -117,7 +117,7 @@ public Request(StreamInput in) throws IOException {
             super(in);
             this.names = in.readStringArray();
             this.indicesOptions = IndicesOptions.readIndicesOptions(in);
-            if (in.getTransportVersion().onOrAfter(TransportVersions.RESOLVE_INDEX_MODE_FILTER)) {
+            if (in.getTransportVersion().supports(RESOLVE_INDEX_MODE_FILTER)) {
                 this.indexModes = in.readEnumSet(IndexMode.class);
             } else {
                 this.indexModes = EnumSet.noneOf(IndexMode.class);
@@ -129,7 +129,7 @@ public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
             out.writeStringArray(names);
             indicesOptions.writeIndicesOptions(out);
-            if (out.getTransportVersion().onOrAfter(TransportVersions.RESOLVE_INDEX_MODE_FILTER)) {
+            if (out.getTransportVersion().supports(RESOLVE_INDEX_MODE_FILTER)) {
                 out.writeEnumSet(indexModes);
             }
         }

server/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java

@@ -11,6 +11,7 @@
 
 import org.apache.lucene.util.Accountable;
 import org.apache.lucene.util.RamUsageEstimator;
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.TransportVersions;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestValidationException;
@@ -65,6 +66,10 @@ public class BulkRequest extends LegacyActionRequest
         Accountable,
         RawIndexingDataTransportRequest {
 
+    private static final TransportVersion STREAMS_ENDPOINT_PARAM_RESTRICTIONS = TransportVersion.fromName(
+        "streams_endpoint_param_restrictions"
+    );
+
     private static final long SHALLOW_SIZE = RamUsageEstimator.shallowSizeOfInstance(BulkRequest.class);
 
     private static final int REQUEST_OVERHEAD = 50;
@@ -110,7 +115,7 @@ public BulkRequest(StreamInput in) throws IOException {
         if (in.getTransportVersion().onOrAfter(TransportVersions.INGEST_REQUEST_INCLUDE_SOURCE_ON_ERROR)) {
             includeSourceOnError = in.readBoolean();
         } // else default value is true
-        if (in.getTransportVersion().onOrAfter(TransportVersions.STREAMS_ENDPOINT_PARAM_RESTRICTIONS)) {
+        if (in.getTransportVersion().supports(STREAMS_ENDPOINT_PARAM_RESTRICTIONS)) {
             paramsUsed = in.readCollectionAsImmutableSet(StreamInput::readString);
         }
     }
@@ -479,7 +484,7 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getTransportVersion().onOrAfter(TransportVersions.INGEST_REQUEST_INCLUDE_SOURCE_ON_ERROR)) {
             out.writeBoolean(includeSourceOnError);
         }
-        if (out.getTransportVersion().onOrAfter(TransportVersions.STREAMS_ENDPOINT_PARAM_RESTRICTIONS)) {
+        if (out.getTransportVersion().supports(STREAMS_ENDPOINT_PARAM_RESTRICTIONS)) {
             out.writeCollection(paramsUsed, StreamOutput::writeString);
         }
     }