[Remote Vector Index Build] Add vector data upload implementation to RemoteIndexBuildStrategy

[jed326]

Description

This PR includes implementation + testing for upload vector and doc id blobs to a remote repository with the RemoteIndexBuildStrategy. This PR also includes a new knn.remote_index_build.size_threshold setting to control the size threshold above which the remote index build service will be used.

---

Vector Upload Details

For a detailed explanation on the vector uploads, please see the Parallel Blob Upload section from the LLD: #2465

At a high level the flow is as follows (Currently only repository-s3 supports parallel blob uploads via this interface, so I will reference S3 directly in the following):

1. KNN Plugin provides an InputStream supplier to asyncBlobUpload (See: RemoteIndexBuildStrategy#getTransferPartStreamSupplier)
2. Repository S3 determines the number of parts (N) to split the blob into based on repository settings. By default we use 16mb parts.
3. Repository uses the supplier from [1] to create N VectorValueInputStreams each representing a only a specific part of the underlying KNNVectorValues.
4. Repository S3 performs a multi-part upload request which each stream as an upload part. The number of parallel parts depends on the upload priority and threadpool configurations through the repository settings.

Because KNNVectorValues is an iterator, we need to create N instances of it, otherwise the iterating through all of the vectors is still done sequentially which will greatly reduce the throughput. VectorValueInputStreams takes in a constructor parameter to position the head of the input stream at a specific byte offset in the KNNVectorValues, including offsets within a given vector. For example, for doc id X the vector could be partially read by InputStream part 1 and partially in InputStream part 2.

---

This PR DOES NOT include:

• Checksum implementation, which will come as a separate follow-up PR to focus on the parallel upload approach here
• Prefix pattern optimizations (see: Optimised prefix pattern per shard for remote store data and metadata files for higher throughput OpenSearch#12567). This will be re-evaluated as a follow-up

For reference, this is the blob path being used:

<BASE_PATH>/<Index UUID>"_vectors"/<uuid>"_"<field name>"_"<segment name".knnvec/knndid

Related Issues

Relates #2465
Relates #2392
Relates #2391

Check List

• New functionality includes testing.
  - [ ] New functionality has been documented.
  - [ ] API changes companion pull request created.
• Commits are signed per the DCO using --signoff.
  - [ ] Public documentation issue/PR created.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[jed326]

@jmazanec15 @shatejas @Vikasht34 @navneet1v could you please help review? thanks!

[navneet1v]

@jed326 can you please add the flow of functions used by the Repository service for InputStream. Since there are some override functions its hard to follow which function is called when.

It will also be good if you can add a packge-info.java explaining the flow and usage of the InputStream in more details.

[navneet1v]

why we need a & here?

[jed326]

We technically don't (added a code comment explaining), however I think it's safer to keep it here. IntelliJ itself also gives you a warning if you don't do the conversion and all the base InputStream implementations do the same.

I also added some tests to verify that read() and read(byte[] b, int offset, int length) read the same data over a given stream.

[navneet1v]

@jed326 lets avoid using skip-change log tag and update changelogs

[jmazanec15]

overall - looks good to me. A few minor comments, but nothing blocking

[jmazanec15]

What makes this time based?

[jed326]

This is the implementation: https://github.com/opensearch-project/OpenSearch/blob/7e2d2437a14580e985440c140f06680dc4e3cd81/libs/common/src/main/java/org/opensearch/common/UUIDs.java#L47-L51

[shatejas]

Looks good overall. A few things related to maintenance and UTs

[shatejas]

Since you already have a testBasePath, can you replace any() with it?

[jed326]

Unfortunately BlobPath path = getRepository().basePath().add(indexSettings.getUUID() + VECTORS_PATH); creates a new BlobPath instances instead of modifying the existing one, so I would need to mock BlobPath itself as well as set up all the mocks for IndexSettings. Since we're using our own custom blobContainer implementation here I don't think it's necessary for this test.

[navneet1v]

Overall code looks good to me. Please fix @shatejas comments specially related to creating repo/dataacess class .

[jed326]

Thanks @navneet1v @shatejas @jmazanec15, I've added a new VectorRepositoryAccessor interface with a default implementation of DefaultVectorRepositoryAccessor to handle all of the KNNVectorValues -> InputStream conversions. Please take another look whenever you get a chance, thanks!

[Vikasht34]

Since we are dealing with call backs and async flow , Better option to use is CompletableFuture than AtomicReference

[jed326]

I looked into this some more, I think the existing implementation is still preferable to CompletableFuture. The reason is asyncBlobUpload does not throw an exception on upload failures, instead it propagates it to whatever ActionListener is passed into it.

If we want to do this with CompletableFuture without an AtomicReference then we would need to throw the exception within the ActionListener, which is not correct behavior as the listener is not intended to interrupt execution flows.

[Vikasht34]

This is how it should work.

CompletableFuture<Void> vectorUploadFuture;
    
    if (blobContainer instanceof AsyncMultiStreamBlobContainer) {
        log.info("Using parallel upload for repository: {}", repository);
        vectorUploadFuture = CompletableFuture.runAsync(() -> {
            try {
                uploadVectorsAsync(blobContainer, blobName, vectorBlobLength, vectorDataType, knnVectorValuesSupplier);
            } catch (IOException e) {
                throw new RuntimeException("Vector upload failed", e);
            }
        });
    } else {
        log.info("Using sequential upload for repository: {}", repository);
        vectorUploadFuture = CompletableFuture.runAsync(() -> {
            try {
                uploadVectorsSequentially(blobContainer, blobName, vectorBlobLength, vectorDataType, knnVectorValues);
            } catch (IOException e) {
                throw new RuntimeException("Sequential vector upload failed", e);
            }
        });
    }

    CompletableFuture<Void> docIdUploadFuture = CompletableFuture.runAsync(() -> {
        try {
            writeDocIds(knnVectorValues, totalLiveDocs, blobName, blobContainer);
        } catch (IOException e) {
            throw new RuntimeException("Doc ID upload failed", e);
        }
    });

    // Wait for both tasks to complete
    CompletableFuture.allOf(vectorUploadFuture, docIdUploadFuture).join();

[Vikasht34]

nit :- Can we have somthing boolean instread of making CurrentBuffer null , like end of stream.

[jed326]

I'm not against using a boolean, but I think it's important to make currentBuffer null to enforce we do not read extra bytes out of the buffer. Then an NPE here would mean a fatal/unrecoverable error in the remote upload process so we want to fail and use the fallbackStrategy. Without setting currentBuffer to null I think we're at risk of reading extra bytes.

[shatejas]

Looks good no major concerns as such

[Vikasht34]

LGTM