-
Notifications
You must be signed in to change notification settings - Fork 1.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Implement block level fetch for Composite Directory
Signed-off-by: Shreyansh Ray <rayshrey@amazon.com>
- Loading branch information
Showing
6 changed files
with
129 additions
and
10 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
93 changes: 93 additions & 0 deletions
93
...rc/main/java/org/opensearch/index/store/remote/file/OnDemandCompositeBlockIndexInput.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,93 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
|
|
||
| package org.opensearch.index.store.remote.file; | ||
|
|
||
| import org.apache.lucene.store.FSDirectory; | ||
| import org.apache.lucene.store.IndexInput; | ||
| import org.opensearch.index.store.CompositeDirectoryRemoteStoreFileTrackerAdapter; | ||
| import org.opensearch.index.store.RemoteStoreFileTrackerAdapter; | ||
| import org.opensearch.index.store.remote.utils.BlobFetchRequest; | ||
|
|
||
| import java.io.IOException; | ||
|
|
||
| public class OnDemandCompositeBlockIndexInput extends OnDemandBlockIndexInput { | ||
|
|
||
| private final RemoteStoreFileTrackerAdapter remoteStoreFileTrackerAdapter; | ||
| private final String fileName; | ||
| private final Long originalFileSize; | ||
| private final FSDirectory directory; | ||
|
|
||
| public OnDemandCompositeBlockIndexInput(RemoteStoreFileTrackerAdapter remoteStoreFileTrackerAdapter, String fileName, FSDirectory directory) { | ||
| this( | ||
| OnDemandBlockIndexInput.builder(). | ||
| resourceDescription("OnDemandCompositeBlockIndexInput"). | ||
| isClone(false). | ||
| offset(0L). | ||
| length(getFileLength(remoteStoreFileTrackerAdapter, fileName)), | ||
| remoteStoreFileTrackerAdapter, | ||
| fileName, | ||
| directory); | ||
| } | ||
|
|
||
| public OnDemandCompositeBlockIndexInput(Builder builder, RemoteStoreFileTrackerAdapter remoteStoreFileTrackerAdapter, String fileName, FSDirectory directory) { | ||
| super(builder); | ||
| this.remoteStoreFileTrackerAdapter = remoteStoreFileTrackerAdapter; | ||
| this.directory = null; | ||
| this.fileName = fileName; | ||
| originalFileSize = getFileLength(remoteStoreFileTrackerAdapter, fileName); | ||
| } | ||
|
|
||
| @Override | ||
| protected OnDemandCompositeBlockIndexInput buildSlice(String sliceDescription, long offset, long length) { | ||
| return new OnDemandCompositeBlockIndexInput( | ||
| OnDemandBlockIndexInput.builder(). | ||
| blockSizeShift(blockSizeShift). | ||
| isClone(true). | ||
| offset(this.offset + offset). | ||
| length(length). | ||
| resourceDescription(sliceDescription), | ||
| remoteStoreFileTrackerAdapter, | ||
| fileName, | ||
| directory | ||
| ); | ||
| } | ||
|
|
||
| @Override | ||
| protected IndexInput fetchBlock(int blockId) throws IOException { | ||
| final String uploadedFileName = ((CompositeDirectoryRemoteStoreFileTrackerAdapter)remoteStoreFileTrackerAdapter).getUploadedFileName(fileName); | ||
| final String blockFileName = uploadedFileName + "." + blockId; | ||
| final long blockStart = getBlockStart(blockId); | ||
| final long length = getActualBlockSize(blockId); | ||
|
|
||
| BlobFetchRequest blobFetchRequest = BlobFetchRequest.builder() | ||
| .position(blockStart) | ||
| .length(length) | ||
| .blobName(uploadedFileName) | ||
| .directory(directory) | ||
| .fileName(blockFileName) | ||
| .build(); | ||
| return remoteStoreFileTrackerAdapter.fetchBlob(blobFetchRequest); | ||
| } | ||
|
|
||
| @Override | ||
| public OnDemandBlockIndexInput clone() { | ||
| OnDemandCompositeBlockIndexInput clone = buildSlice("clone", 0L, this.length); | ||
| // ensures that clones may be positioned at the same point as the blocked file they were cloned from | ||
| clone.cloneBlock(this); | ||
| return clone; | ||
| } | ||
|
|
||
| private long getActualBlockSize(int blockId) { | ||
| return (blockId != getBlock(originalFileSize - 1)) ? blockSize : getBlockOffset(originalFileSize - 1) + 1; | ||
| } | ||
|
|
||
| private static long getFileLength(RemoteStoreFileTrackerAdapter remoteStoreFileTrackerAdapter, String fileName) { | ||
| return ((CompositeDirectoryRemoteStoreFileTrackerAdapter)remoteStoreFileTrackerAdapter).getFileLength(fileName); | ||
| } | ||
| } |