add reconsider_merge_policy to directory

src/directory/directory.rs

@@ -2,6 +2,7 @@ use crate::directory::directory_lock::Lock;
 use crate::directory::error::{DeleteError, LockError, OpenReadError, OpenWriteError};
 use crate::directory::{FileHandle, FileSlice, WatchCallback, WatchHandle, WritePtr};
 use crate::index::SegmentMetaInventory;
+use crate::merge_policy::MergePolicy;
 use crate::IndexMeta;
 use std::any::Any;
 use std::collections::HashSet;
@@ -265,6 +266,16 @@ pub trait Directory: DirectoryClone + fmt::Debug + Send + Sync + 'static {
             "load_metas not implemented".to_string(),
         ))
     }
+
+    // Allows the directory to change the writer's merge policy right before the merge happens
+    // This is useful for directories that need to change the merge policy based on how many segments were created
+    fn reconsider_merge_policy(
+        &self,
+        _metas: &IndexMeta,
+        _previous_metas: &IndexMeta,
+    ) -> Option<Box<dyn MergePolicy>> {
+        None
+    }
 }
 
 /// DirectoryClone

src/directory/managed_directory.rs

@@ -13,6 +13,7 @@ use crate::directory::{
     DirectoryLock, FileHandle, FileSlice, GarbageCollectionResult, Lock, WatchCallback,
     WatchHandle, WritePtr, MANAGED_LOCK, META_LOCK,
 };
+use crate::merge_policy::MergePolicy;
 use crate::error::DataCorruption;
 use crate::index::SegmentMetaInventory;
 use crate::{Directory, IndexMeta};
@@ -349,6 +350,14 @@ impl Directory for ManagedDirectory {
     fn load_metas(&self, inventory: &SegmentMetaInventory) -> crate::Result<IndexMeta> {
         self.directory.load_metas(inventory)
     }
+
+    fn reconsider_merge_policy(
+        &self,
+        metas: &IndexMeta,
+        previous_metas: &IndexMeta,
+    ) -> Option<Box<dyn MergePolicy>> {
+        self.directory.reconsider_merge_policy(metas, previous_metas)
+    }
 }
 
 impl Clone for ManagedDirectory {

src/indexer/segment_updater.rs

@@ -466,6 +466,16 @@ impl SegmentUpdater {
             segment_updater.segment_manager.commit(segment_entries);
             segment_updater.save_metas(opstamp, payload, &previous_metas)?;
             let _ = garbage_collect_files(segment_updater.clone());
+
+            let index_meta = segment_updater.load_meta();
+            if let Some(new_merge_policy) = segment_updater
+                .index
+                .directory()
+                .reconsider_merge_policy(&index_meta, &previous_metas)
+            {
+                segment_updater.set_merge_policy(new_merge_policy);
+            }
+
             segment_updater.consider_merge_options();
             Ok(opstamp)
         })