[POC] primary term based directory structure

Signed-off-by: Varun Bansal <bansvaru@amazon.com>

Author: linuxpi

server/src/main/java/org/opensearch/index/shard/IndexShard.java

@@ -184,6 +184,7 @@
 import org.opensearch.index.translog.RemoteFsTranslog;
 import org.opensearch.index.translog.RemoteTranslogStats;
 import org.opensearch.index.translog.Translog;
+import org.opensearch.index.translog.Translog.Durability;
 import org.opensearch.index.translog.TranslogConfig;
 import org.opensearch.index.translog.TranslogFactory;
 import org.opensearch.index.translog.TranslogRecoveryRunner;
@@ -569,6 +570,32 @@ private long getInitialGlobalCheckpointForShard(IndexSettings indexSettings) {
         return UNASSIGNED_SEQ_NO;
     }
 
+    /**
+     * Initializes primary term routing for the store directory if supported.
+     * This method should be called after the IndexShard is fully constructed
+     * to enable primary term-based segment file routing.
+     */
+    private void initializePrimaryTermRouting() {
+        logger.info("intializing primary term based routing");
+        try {
+            Directory storeDirectory = store.directory();
+            logger.info("intializing primary term based routing {}", storeDirectory);
+            // if (storeDirectory instanceof org.opensearch.index.store.distributed.PrimaryTermAwareDirectoryWrapper) {
+                org.opensearch.index.store.distributed.PrimaryTermAwareDirectoryWrapper wrapper = 
+                    (org.opensearch.index.store.distributed.PrimaryTermAwareDirectoryWrapper) ((FilterDirectory) ((FilterDirectory)storeDirectory).getDelegate()).getDelegate();
+                logger.info("intializing primary term based routing");
+                
+                if (!wrapper.isPrimaryTermRoutingEnabled()) {
+                    wrapper.enablePrimaryTermRouting(this);
+                    logger.info("Enabled primary term routing for shard {}", shardId);
+                }
+            // }
+        } catch (Exception e) {
+            logger.warn("Failed to initialize primary term routing for shard {}", shardId, e);
+            // Don't fail shard creation if primary term routing setup fails
+        }
+    }
+
     public ThreadPool getThreadPool() {
         return this.threadPool;
     }
@@ -2444,6 +2471,9 @@ public void postRecovery(String reason) throws IndexShardStartedException, Index
                 }
                 recoveryState.setStage(RecoveryState.Stage.DONE);
                 changeState(IndexShardState.POST_RECOVERY, reason);
+                
+                // Initialize primary term routing after recovery is complete
+                initializePrimaryTermRouting();
             }
         }
     }

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

@@ -50,6 +50,7 @@
 import org.opensearch.index.IndexSettings;
 import org.opensearch.index.shard.ShardPath;
 import org.opensearch.index.store.distributed.DistributedSegmentDirectory;
+import org.opensearch.index.store.distributed.PrimaryTermAwareDirectoryWrapper;
 import org.opensearch.plugins.IndexStorePlugin;
 
 import java.io.IOException;
@@ -97,10 +98,21 @@ protected Directory newFSDirectory(Path location, LockFactory lockFactory, Index
         Set<String> preLoadExtensions = new HashSet<>(indexSettings.getValue(IndexModule.INDEX_STORE_PRE_LOAD_SETTING));
         switch (type) {
             case HYBRIDFS:
-                // return new DistributedSegmentDirectory(null, location)
-                // Use Lucene defaults
+                // Create primary directory
                 final FSDirectory primaryDirectory = new NIOFSDirectory(location, lockFactory);
-                return new DistributedSegmentDirectory(primaryDirectory, location);
+                
+                return new PrimaryTermAwareDirectoryWrapper(primaryDirectory, location);
+                // // Check if primary term routing should be enabled
+                // boolean enablePrimaryTermRouting = indexSettings.getSettings()
+                //     .getAsBoolean("index.store.distributed_segment.enable_primary_term_routing", true);
+                
+                // if (enablePrimaryTermRouting) {
+                //     // Use wrapper that can be configured for primary term routing later
+                //     return new PrimaryTermAwareDirectoryWrapper(primaryDirectory, location);
+                // } else {
+                //     // Use legacy hash-based routing
+                //     return new DistributedSegmentDirectory(primaryDirectory, location);
+                // }
 
                 // final Set<String> nioExtensions = new HashSet<>(indexSettings.getValue(IndexModule.INDEX_STORE_HYBRID_NIO_EXTENSIONS));
                 // if (primaryDirectory instanceof MMapDirectory) {