Skip to content

Commit 79c1656

Browse files
linuxpilinuxpi
committed
[POC] supporting nested directory structure for single shard data
Signed-off-by: Varun Bansal <bansvaru@amazon.com>
1 parent 432d7d1 commit 79c1656

17 files changed

Lines changed: 3190 additions & 8 deletions

‎server/src/main/java/org/opensearch/index/store/FsDirectoryFactory.java‎

Lines changed: 12 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@
4949
import org.opensearch.index.IndexModule;
5050
import org.opensearch.index.IndexSettings;
5151
import org.opensearch.index.shard.ShardPath;
52+
import org.opensearch.index.store.distributed.DistributedSegmentDirectory;
5253
import org.opensearch.plugins.IndexStorePlugin;
5354

5455
import java.io.IOException;
@@ -96,15 +97,18 @@ protected Directory newFSDirectory(Path location, LockFactory lockFactory, Index
9697
Set<String> preLoadExtensions = new HashSet<>(indexSettings.getValue(IndexModule.INDEX_STORE_PRE_LOAD_SETTING));
9798
switch (type) {
9899
case HYBRIDFS:
100+
// return new DistributedSegmentDirectory(null, location)
99101
// Use Lucene defaults
100-
final FSDirectory primaryDirectory = FSDirectory.open(location, lockFactory);
101-
final Set<String> nioExtensions = new HashSet<>(indexSettings.getValue(IndexModule.INDEX_STORE_HYBRID_NIO_EXTENSIONS));
102-
if (primaryDirectory instanceof MMapDirectory) {
103-
MMapDirectory mMapDirectory = (MMapDirectory) primaryDirectory;
104-
return new HybridDirectory(lockFactory, setPreload(mMapDirectory, preLoadExtensions), nioExtensions);
105-
} else {
106-
return primaryDirectory;
107-
}
102+
final FSDirectory primaryDirectory = new NIOFSDirectory(location, lockFactory);
103+
return new DistributedSegmentDirectory(primaryDirectory, location);
104+
105+
// final Set<String> nioExtensions = new HashSet<>(indexSettings.getValue(IndexModule.INDEX_STORE_HYBRID_NIO_EXTENSIONS));
106+
// if (primaryDirectory instanceof MMapDirectory) {
107+
// MMapDirectory mMapDirectory = (MMapDirectory) primaryDirectory;
108+
// return new HybridDirectory(lockFactory, setPreload(mMapDirectory, preLoadExtensions), nioExtensions);
109+
// } else {
110+
// return primaryDirectory;
111+
// }
108112
case MMAPFS:
109113
return setPreload(new MMapDirectory(location, lockFactory), preLoadExtensions);
110114
// simplefs was removed in Lucene 9; support for enum is maintained for bwc

‎server/src/main/java/org/opensearch/index/store/distributed/DefaultFilenameHasher.java‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
*
4+
* The OpenSearch Contributors require contributions made to
5+
* this file be licensed under the Apache-2.0 license or a
6+
* compatible open source license.
7+
*/
8+
9+
package org.opensearch.index.store.distributed;
10+
11+
import java.util.Set;
12+
13+
/**
14+
* Default implementation of FilenameHasher that uses consistent hashing
15+
* to distribute segment files across multiple directories while keeping
16+
* critical files like segments_N in the base directory.
17+
*
18+
* @opensearch.internal
19+
*/
20+
public class DefaultFilenameHasher implements FilenameHasher {
21+
22+
private static final int NUM_DIRECTORIES = 5;
23+
private static final Set<String> EXCLUDED_PREFIXES = Set.of("segments_", "pending_segments_", "write.lock");
24+
25+
/**
26+
* Maps a filename to a directory index using consistent hashing.
27+
* Files with excluded prefixes (like segments_N) are always mapped to index 0.
28+
*
29+
* @param filename the segment filename to hash
30+
* @return directory index between 0 and 4 (inclusive)
31+
* @throws IllegalArgumentException if filename is null or empty
32+
*/
33+
@Override
34+
public int getDirectoryIndex(String filename) {
35+
if (filename == null || filename.isEmpty()) {
36+
throw new IllegalArgumentException("Filename cannot be null or empty");
37+
}
38+
39+
if (isExcludedFile(filename)) {
40+
return 0; // Base directory for excluded files
41+
}
42+
43+
// Use consistent hashing with absolute value to ensure positive result
44+
return Math.abs(filename.hashCode()) % NUM_DIRECTORIES;
45+
}
46+
47+
/**
48+
* Checks if a filename should be excluded from distribution.
49+
* Currently excludes files starting with "segments_" prefix.
50+
*
51+
* @param filename the filename to check
52+
* @return true if file should remain in base directory, false if it can be distributed
53+
*/
54+
@Override
55+
public boolean isExcludedFile(String filename) {
56+
if (filename == null || filename.isEmpty()) {
57+
return true; // Treat invalid filenames as excluded
58+
}
59+
60+
if (filename.endsWith(".tmp")) {
61+
return true;
62+
}
63+
64+
return EXCLUDED_PREFIXES.stream().anyMatch(filename::startsWith);
65+
}
66+
}

‎server/src/main/java/org/opensearch/index/store/distributed/DirectoryManager.java‎

Lines changed: 163 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,163 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
*
4+
* The OpenSearch Contributors require contributions made to
5+
* this file be licensed under the Apache-2.0 license or a
6+
* compatible open source license.
7+
*/
8+
9+
package org.opensearch.index.store.distributed;
10+
11+
import org.apache.lucene.store.Directory;
12+
import org.apache.lucene.store.FSDirectory;
13+
import org.apache.lucene.store.FSLockFactory;
14+
import org.apache.lucene.store.NIOFSDirectory;
15+
16+
import java.io.IOException;
17+
import java.nio.file.Files;
18+
import java.nio.file.Path;
19+
20+
/**
21+
* Manages the creation and access to subdirectories for distributed segment storage.
22+
* Creates up to 5 subdirectories for distributing segment files while keeping
23+
* the base directory (index 0) for critical files like segments_N.
24+
*
25+
* @opensearch.internal
26+
*/
27+
public class DirectoryManager {
28+
29+
private static final int NUM_DIRECTORIES = 5;
30+
private final Path baseDirectory;
31+
private final Directory[] subdirectories;
32+
33+
/**
34+
* Creates a new DirectoryManager with the specified base directory.
35+
*
36+
* @param baseDirectory the base Directory instance (used as subdirectory 0)
37+
* @param basePath the base filesystem path for creating subdirectories
38+
* @throws IOException if subdirectory creation fails
39+
*/
40+
public DirectoryManager(Directory baseDirectory, Path basePath) throws IOException {
41+
this.baseDirectory = basePath;
42+
this.subdirectories = createSubdirectories(baseDirectory, basePath);
43+
}
44+
45+
/**
46+
* Creates the array of subdirectories, with index 0 being the base directory
47+
* and indices 1-4 being newly created subdirectories.
48+
*
49+
* @param base the base Directory instance
50+
* @param basePath the base filesystem path
51+
* @return array of Directory instances
52+
* @throws IOException if subdirectory creation fails
53+
*/
54+
private Directory[] createSubdirectories(Directory base, Path basePath) throws IOException {
55+
Directory[] dirs = new Directory[NUM_DIRECTORIES];
56+
dirs[0] = base; // Base directory for segments_N and excluded files
57+
58+
try {
59+
for (int i = 1; i < NUM_DIRECTORIES; i++) {
60+
Path subPath = basePath.resolve("varun_segments_" + i);
61+
62+
// Create directory if it doesn't exist
63+
if (!Files.exists(subPath)) {
64+
Files.createDirectories(subPath);
65+
}
66+
67+
// Validate directory is writable
68+
if (!Files.isWritable(subPath)) {
69+
throw new IOException("Subdirectory is not writable: " + subPath);
70+
}
71+
72+
dirs[i] = new NIOFSDirectory(subPath, FSLockFactory.getDefault());
73+
}
74+
} catch (IOException e) {
75+
// Clean up any successfully created directories
76+
closeDirectories(dirs);
77+
throw new DistributedDirectoryException(
78+
"Failed to create subdirectories",
79+
-1,
80+
"createSubdirectories",
81+
e
82+
);
83+
}
84+
85+
return dirs;
86+
}
87+
88+
/**
89+
* Gets the Directory instance for the specified index.
90+
*
91+
* @param index the directory index (0-4)
92+
* @return the Directory instance
93+
* @throws IllegalArgumentException if index is out of range
94+
*/
95+
public Directory getDirectory(int index) {
96+
if (index < 0 || index >= NUM_DIRECTORIES) {
97+
throw new IllegalArgumentException("Directory index must be between 0 and " + (NUM_DIRECTORIES - 1) +
98+
", got: " + index);
99+
}
100+
return subdirectories[index];
101+
}
102+
103+
/**
104+
* Gets the number of managed directories.
105+
*
106+
* @return the number of directories (always 5)
107+
*/
108+
public int getNumDirectories() {
109+
return NUM_DIRECTORIES;
110+
}
111+
112+
/**
113+
* Gets the base filesystem path.
114+
*
115+
* @return the base path
116+
*/
117+
public Path getBasePath() {
118+
return baseDirectory;
119+
}
120+
121+
/**
122+
* Closes all managed directories except the base directory (index 0).
123+
* The base directory should be closed by the caller since it was provided
124+
* during construction.
125+
*
126+
* @throws IOException if any directory fails to close
127+
*/
128+
public void close() throws IOException {
129+
closeDirectories(subdirectories);
130+
}
131+
132+
/**
133+
* Helper method to close directories and collect exceptions.
134+
*
135+
* @param dirs array of directories to close
136+
* @throws IOException if any directory fails to close
137+
*/
138+
private void closeDirectories(Directory[] dirs) throws IOException {
139+
IOException exception = null;
140+
141+
// Close subdirectories (skip index 0 as it's the base directory managed externally)
142+
for (int i = 1; i < dirs.length && dirs[i] != null; i++) {
143+
try {
144+
dirs[i].close();
145+
} catch (IOException e) {
146+
if (exception == null) {
147+
exception = new DistributedDirectoryException(
148+
"Failed to close subdirectory",
149+
i,
150+
"close",
151+
e
152+
);
153+
} else {
154+
exception.addSuppressed(e);
155+
}
156+
}
157+
}
158+
159+
if (exception != null) {
160+
throw exception;
161+
}
162+
}
163+
}

‎server/src/main/java/org/opensearch/index/store/distributed/DistributedDirectoryException.java‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,67 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
*
4+
* The OpenSearch Contributors require contributions made to
5+
* this file be licensed under the Apache-2.0 license or a
6+
* compatible open source license.
7+
*/
8+
9+
package org.opensearch.index.store.distributed;
10+
11+
import java.io.IOException;
12+
13+
/**
14+
* Exception thrown when operations on a distributed segment directory fail.
15+
* This exception provides additional context about which directory index
16+
* and operation caused the failure to aid in debugging and monitoring.
17+
*
18+
* @opensearch.internal
19+
*/
20+
public class DistributedDirectoryException extends IOException {
21+
22+
private final int directoryIndex;
23+
private final String operation;
24+
25+
/**
26+
* Creates a new DistributedDirectoryException with directory context.
27+
*
28+
* @param message the error message
29+
* @param directoryIndex the index of the directory where the error occurred (0-4)
30+
* @param operation the operation that was being performed when the error occurred
31+
* @param cause the underlying cause of the exception
32+
*/
33+
public DistributedDirectoryException(String message, int directoryIndex, String operation, Throwable cause) {
34+
super(String.format("Directory %d operation '%s' failed: %s", directoryIndex, operation, message), cause);
35+
this.directoryIndex = directoryIndex;
36+
this.operation = operation;
37+
}
38+
39+
/**
40+
* Creates a new DistributedDirectoryException with directory context.
41+
*
42+
* @param message the error message
43+
* @param directoryIndex the index of the directory where the error occurred (0-4)
44+
* @param operation the operation that was being performed when the error occurred
45+
*/
46+
public DistributedDirectoryException(String message, int directoryIndex, String operation) {
47+
this(message, directoryIndex, operation, null);
48+
}
49+
50+
/**
51+
* Gets the directory index where the error occurred.
52+
*
53+
* @return the directory index (0-4)
54+
*/
55+
public int getDirectoryIndex() {
56+
return directoryIndex;
57+
}
58+
59+
/**
60+
* Gets the operation that was being performed when the error occurred.
61+
*
62+
* @return the operation name
63+
*/
64+
public String getOperation() {
65+
return operation;
66+
}
67+
}

0 commit comments

Comments
 (0)