Merge branch 'main' into t-digest-sum-of-empty-change

Author: not-napoleon

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3BlobContainer.java

@@ -783,7 +783,10 @@ static Tuple<Long, Long> numberOfMultiparts(final long totalSize, final long par
         }
     }
 
-    private class CompareAndExchangeOperation {
+    /**
+     * An implementation of {@link BlobContainer#compareAndExchangeRegister} based on strongly-consistent multipart upload APIs.
+     */
+    private class MultipartUploadCompareAndExchangeOperation {
 
         private final OperationPurpose purpose;
         private final S3Client client;
@@ -792,7 +795,13 @@ private class CompareAndExchangeOperation {
         private final String blobKey;
         private final ThreadPool threadPool;
 
-        CompareAndExchangeOperation(OperationPurpose purpose, S3Client client, String bucket, String key, ThreadPool threadPool) {
+        MultipartUploadCompareAndExchangeOperation(
+            OperationPurpose purpose,
+            S3Client client,
+            String bucket,
+            String key,
+            ThreadPool threadPool
+        ) {
             this.purpose = purpose;
             this.client = client;
             this.bucket = bucket;
@@ -802,6 +811,23 @@ private class CompareAndExchangeOperation {
         }
 
         void run(BytesReference expected, BytesReference updated, ActionListener<OptionalBytesReference> listener) throws Exception {
+            innerRun(expected, updated, listener.delegateResponse((delegate, e) -> {
+                logger.trace(() -> Strings.format("[%s]: compareAndExchangeRegister failed", rawKey), e);
+                if (e instanceof AwsServiceException awsServiceException
+                    && (awsServiceException.statusCode() == 404
+                        || awsServiceException.statusCode() == 200
+                            && "NoSuchUpload".equals(awsServiceException.awsErrorDetails().errorCode()))) {
+                    // An uncaught 404 means that our multipart upload was aborted by a concurrent operation before we could complete it.
+                    // Also (rarely) S3 can start processing the request during a concurrent abort and this can result in a 200 OK with an
+                    // <Error><Code>NoSuchUpload</Code>... in the response. Either way, this means that our write encountered contention:
+                    delegate.onResponse(OptionalBytesReference.MISSING);
+                } else {
+                    delegate.onFailure(e);
+                }
+            }));
+        }
+
+        void innerRun(BytesReference expected, BytesReference updated, ActionListener<OptionalBytesReference> listener) throws Exception {
             BlobContainerUtils.ensureValidRegisterContent(updated);
 
             if (hasPreexistingUploads()) {
@@ -1094,25 +1120,15 @@ public void compareAndExchangeRegister(
         ActionListener<OptionalBytesReference> listener
     ) {
         final var clientReference = blobStore.clientReference();
-        ActionListener.run(ActionListener.releaseAfter(listener.delegateResponse((delegate, e) -> {
-            logger.trace(() -> Strings.format("[%s]: compareAndExchangeRegister failed", key), e);
-            if (e instanceof AwsServiceException awsServiceException
-                && (awsServiceException.statusCode() == 404
-                    || awsServiceException.statusCode() == 200
-                        && "NoSuchUpload".equals(awsServiceException.awsErrorDetails().errorCode()))) {
-                // An uncaught 404 means that our multipart upload was aborted by a concurrent operation before we could complete it.
-                // Also (rarely) S3 can start processing the request during a concurrent abort and this can result in a 200 OK with an
-                // <Error><Code>NoSuchUpload</Code>... in the response. Either way, this means that our write encountered contention:
-                delegate.onResponse(OptionalBytesReference.MISSING);
-            } else {
-                delegate.onFailure(e);
-            }
-        }), clientReference),
-            l -> new CompareAndExchangeOperation(purpose, clientReference.client(), blobStore.bucket(), key, blobStore.getThreadPool()).run(
-                expected,
-                updated,
-                l
-            )
+        ActionListener.run(
+            ActionListener.releaseBefore(clientReference, listener),
+            l -> new MultipartUploadCompareAndExchangeOperation(
+                purpose,
+                clientReference.client(),
+                blobStore.bucket(),
+                key,
+                blobStore.getThreadPool()
+            ).run(expected, updated, l)
         );
     }
 

server/src/main/java/org/elasticsearch/index/codec/vectors/diskbbq/next/ESNextDiskBBQVectorsReader.java

@@ -324,6 +324,20 @@ private static CentroidIterator getCentroidIteratorWithParents(
         final NeighborQueue currentParentQueue = new NeighborQueue(maxChildrenSize, true);
         final int bufferSize = (int) Math.min(Math.max(centroidRatio * numCentroids, 1), numCentroids);
         final int numCentroidsFiltered = acceptCentroids == null ? numCentroids : acceptCentroids.cardinality();
+        if (numCentroidsFiltered == 0) {
+            // TODO maybe this makes CentroidIterator polymorphic?
+            return new CentroidIterator() {
+                @Override
+                public boolean hasNext() {
+                    return false;
+                }
+
+                @Override
+                public CentroidOffsetAndLength nextPostingListOffsetAndLength() {
+                    return null;
+                }
+            };
+        }
         final float[] scores = new float[ES92Int7VectorsScorer.BULK_SIZE];
         final NeighborQueue neighborQueue;
         if (acceptCentroids != null && numCentroidsFiltered <= bufferSize) {

server/src/test/java/org/elasticsearch/index/codec/vectors/diskbbq/next/ESNextDiskBBQVectorsFormatTests.java

@@ -29,6 +29,7 @@
 import org.apache.lucene.index.VectorEncoding;
 import org.apache.lucene.index.VectorSimilarityFunction;
 import org.apache.lucene.search.AcceptDocs;
+import org.apache.lucene.search.DocIdSetIterator;
 import org.apache.lucene.search.KnnCollector;
 import org.apache.lucene.search.TopDocs;
 import org.apache.lucene.search.TopKnnCollector;
@@ -382,6 +383,13 @@ private void doRestrictiveFilter(boolean dense) throws IOException {
                     uniqueDocIds.add(topDocs.scoreDocs[i].doc);
                 }
                 assertEquals(matchingDocs, uniqueDocIds.size());
+                // match no docs
+                leafReader.searchNearestVectors(
+                    "f",
+                    vector,
+                    new TopKnnCollector(2, Integer.MAX_VALUE),
+                    AcceptDocs.fromIteratorSupplier(DocIdSetIterator::empty, leafReader.getLiveDocs(), leafReader.maxDoc())
+                );
             }
         }
     }

x-pack/plugin/snapshot-repo-test-kit/qa/azure/src/javaRestTest/java/org/elasticsearch/repositories/blobstore/testkit/analyze/AzureRepositoryAnalysisRestIT.java

@@ -74,6 +74,7 @@ private static Predicate<String> decideAuthHeaderPredicate() {
     private static final ElasticsearchCluster cluster = ElasticsearchCluster.local()
         .module("repository-azure")
         .module("snapshot-repo-test-kit")
+        .setting("thread_pool.snapshot.max", "10")
         .keystore("azure.client.repository_test_kit.account", AZURE_TEST_ACCOUNT)
         .keystore("azure.client.repository_test_kit.key", () -> AZURE_TEST_KEY, s -> Strings.hasText(AZURE_TEST_KEY))
         .keystore("azure.client.repository_test_kit.sas_token", () -> AZURE_TEST_SASTOKEN, s -> Strings.hasText(AZURE_TEST_SASTOKEN))

x-pack/plugin/snapshot-repo-test-kit/qa/gcs/src/javaRestTest/java/org/elasticsearch/repositories/blobstore/testkit/analyze/GCSRepositoryAnalysisRestIT.java

@@ -31,6 +31,7 @@ public class GCSRepositoryAnalysisRestIT extends AbstractRepositoryAnalysisRestT
     private static ElasticsearchCluster cluster = ElasticsearchCluster.local()
         .module("repository-gcs")
         .module("snapshot-repo-test-kit")
+        .setting("thread_pool.snapshot.max", "10")
         .setting("gcs.client.repository_test_kit.endpoint", () -> fixture.getAddress(), s -> USE_FIXTURE)
         .setting("gcs.client.repository_test_kit.token_uri", () -> fixture.getAddress() + "/o/oauth2/token", s -> USE_FIXTURE)
         .apply(c -> {

x-pack/plugin/snapshot-repo-test-kit/qa/hdfs/src/javaRestTest/java/org/elasticsearch/repositories/blobstore/testkit/analyze/HdfsRepositoryAnalysisRestIT.java

@@ -24,6 +24,7 @@ public class HdfsRepositoryAnalysisRestIT extends AbstractHdfsRepositoryAnalysis
     public static ElasticsearchCluster cluster = ElasticsearchCluster.local()
         .distribution(DistributionType.DEFAULT)
         .plugin("repository-hdfs")
+        .setting("thread_pool.snapshot.max", "10")
         .setting("xpack.license.self_generated.type", "trial")
         .setting("xpack.security.enabled", "false")
         .build();

x-pack/plugin/snapshot-repo-test-kit/qa/hdfs/src/javaRestTest/java/org/elasticsearch/repositories/blobstore/testkit/analyze/SecureHdfsRepositoryAnalysisRestIT.java

@@ -31,6 +31,7 @@ public class SecureHdfsRepositoryAnalysisRestIT extends AbstractHdfsRepositoryAn
     public static ElasticsearchCluster cluster = ElasticsearchCluster.local()
         .distribution(DistributionType.DEFAULT)
         .plugin("repository-hdfs")
+        .setting("thread_pool.snapshot.max", "10")
         .setting("xpack.license.self_generated.type", "trial")
         .setting("xpack.security.enabled", "false")
         .systemProperty("java.security.krb5.conf", () -> krb5Fixture.getConfPath().toString())

x-pack/plugin/snapshot-repo-test-kit/qa/minio/src/javaRestTest/java/org/elasticsearch/repositories/blobstore/testkit/analyze/MinioRepositoryAnalysisRestIT.java

@@ -31,6 +31,7 @@ public class MinioRepositoryAnalysisRestIT extends AbstractRepositoryAnalysisRes
         .distribution(DistributionType.DEFAULT)
         .keystore("s3.client.repository_test_kit.access_key", "s3_test_access_key")
         .keystore("s3.client.repository_test_kit.secret_key", "s3_test_secret_key")
+        .setting("thread_pool.snapshot.max", "10")
         .setting("s3.client.repository_test_kit.endpoint", minioFixture::getAddress)
         .setting("xpack.security.enabled", "false")
         .setting("xpack.ml.enabled", "false")

x-pack/plugin/snapshot-repo-test-kit/qa/s3/src/javaRestTest/java/org/elasticsearch/repositories/blobstore/testkit/analyze/S3RepositoryAnalysisRestIT.java

@@ -82,6 +82,7 @@ private static void ensurePurposeParameterPresent(S3HttpHandler.S3Request reques
         .setting("s3.client.repo_test_kit.region", regionSupplier, (n) -> USE_FIXTURE)
         .setting("s3.client.repo-test_kit.add_purpose_custom_query_parameter", () -> randomFrom("true", "false"), n -> randomBoolean())
         .setting("xpack.security.enabled", "false")
+        .setting("thread_pool.snapshot.max", "10")
         .build();
 
     @ClassRule