Support multi-tenant RAM buffers for IndexWriter

lucene/backward-codecs/src/test/org/apache/lucene/backward_index/TestIndexUpgradeBackwardsCompatibility.java

@@ -26,6 +26,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
+import org.apache.lucene.analysis.Analyzer;
 import org.apache.lucene.document.Document;
 import org.apache.lucene.index.DirectoryReader;
 import org.apache.lucene.index.IndexUpgrader;
@@ -131,7 +132,7 @@ public void testUpgradeOldSingleSegmentIndexWithAdditions() throws Exception {
     // add dummy segments (which are all in current
     // version) to single segment index
     MergePolicy mp = random().nextBoolean() ? newLogMergePolicy() : newTieredMergePolicy();
-    IndexWriterConfig iwc = new IndexWriterConfig(null).setMergePolicy(mp);
+    IndexWriterConfig iwc = new IndexWriterConfig((Analyzer) null).setMergePolicy(mp);
     IndexWriter w = new IndexWriter(directory, iwc);
     w.addIndexes(ramDir);
     try (w) {

lucene/benchmark/src/test/org/apache/lucene/benchmark/byTask/tasks/TestAddIndexesTask.java

@@ -18,6 +18,7 @@
 
 import java.nio.file.Path;
 import java.util.Properties;
+import org.apache.lucene.analysis.Analyzer;
 import org.apache.lucene.benchmark.BenchmarkTestCase;
 import org.apache.lucene.benchmark.byTask.PerfRunData;
 import org.apache.lucene.benchmark.byTask.utils.Config;
@@ -45,7 +46,7 @@ public static void beforeClassAddIndexesTaskTest() throws Exception {
     inputDir = testDir.resolve("input");
     Directory tmpDir = newFSDirectory(inputDir);
     try {
-      IndexWriter writer = new IndexWriter(tmpDir, new IndexWriterConfig(null));
+      IndexWriter writer = new IndexWriter(tmpDir, new IndexWriterConfig((Analyzer) null));
       for (int i = 0; i < 10; i++) {
         writer.addDocument(new Document());
       }

lucene/core/src/java/org/apache/lucene/index/FlushByRamOrCountsPolicy.java

@@ -16,6 +16,8 @@
  */
 package org.apache.lucene.index;
 
+import java.io.IOException;
+
 /**
  * Default {@link FlushPolicy} implementation that flushes new segments based on RAM used and
  * document count depending on the IndexWriter's {@link IndexWriterConfig}. It also applies pending
@@ -52,6 +54,17 @@ && flushOnDocCount()
     }
   }
 
+  @Override
+  public void flushWriter(
+      IndexWriterRAMManager ramManager,
+      IndexWriterRAMManager.PerWriterIndexWriterRAMManager perWriterRamManager)
+      throws IOException {
+    long totalBytes = perWriterRamManager.getTotalBufferBytesUsed();
+    if (totalBytes > ramManager.getRamBufferSizeMB() * 1024 * 1024) {
+      ramManager.flushRoundRobin();
+    }
+  }
+
   private void flushDeletes(DocumentsWriterFlushControl control) {
     control.setApplyAllDeletes();
     if (infoStream.isEnabled("FP")) {

lucene/core/src/java/org/apache/lucene/index/FlushPolicy.java

@@ -16,6 +16,7 @@
  */
 package org.apache.lucene.index;
 
+import java.io.IOException;
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.util.InfoStream;
 
@@ -57,6 +58,17 @@ abstract class FlushPolicy {
   public abstract void onChange(
       DocumentsWriterFlushControl control, DocumentsWriterPerThread perThread);
 
+  /**
+   * Chooses which writer should be flushed. Default implementation chooses the writer with most RAM
+   * usage
+   *
+   * @param ramManager the {@link IndexWriterRAMManager} being used to actually flush the writers
+   */
+  public abstract void flushWriter(
+      IndexWriterRAMManager ramManager,
+      IndexWriterRAMManager.PerWriterIndexWriterRAMManager perWriterRamManager)
+      throws IOException;
+
   /** Called by DocumentsWriter to initialize the FlushPolicy */
   protected synchronized void init(LiveIndexWriterConfig indexWriterConfig) {
     this.indexWriterConfig = indexWriterConfig;

lucene/core/src/java/org/apache/lucene/index/IndexUpgrader.java

@@ -20,6 +20,7 @@
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.Collection;
+import org.apache.lucene.analysis.Analyzer;
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.store.FSDirectory;
 import org.apache.lucene.util.CommandLineUtil;
@@ -135,7 +136,7 @@ static IndexUpgrader parseArgs(String[] args) throws IOException {
    * {@code matchVersion}. The tool refuses to upgrade indexes with multiple commit points.
    */
   public IndexUpgrader(Directory dir) {
-    this(dir, new IndexWriterConfig(null), false);
+    this(dir, new IndexWriterConfig((Analyzer) null), false);
   }
 
   /**
@@ -145,7 +146,7 @@ public IndexUpgrader(Directory dir) {
    * be sent to this stream.
    */
   public IndexUpgrader(Directory dir, InfoStream infoStream, boolean deletePriorCommits) {
-    this(dir, new IndexWriterConfig(null), deletePriorCommits);
+    this(dir, new IndexWriterConfig((Analyzer) null), deletePriorCommits);
     if (null != infoStream) {
       this.iwc.setInfoStream(infoStream);
     }

lucene/core/src/java/org/apache/lucene/index/IndexWriter.java

@@ -462,6 +462,8 @@ public void onTicketBacklog() {
         }
       };
 
+  private final IndexWriterRAMManager.PerWriterIndexWriterRAMManager indexWriterRAMManager;
+
   /**
    * Expert: returns a readonly reader, covering all committed as well as un-committed changes to
    * the index. This provides "near real-time" searching, in that changes made during an IndexWriter
@@ -1211,6 +1213,9 @@ public IndexWriter(Directory d, IndexWriterConfig conf) throws IOException {
         writeLock = null;
       }
     }
+    this.indexWriterRAMManager =
+        new IndexWriterRAMManager.PerWriterIndexWriterRAMManager(
+            this, config.getIndexWriterRAMManager());
   }
 
   /** Confirms that the incoming index sort (if any) matches the existing index sort (if any). */
@@ -1365,6 +1370,7 @@ private void shutdown() throws IOException {
    */
   @Override
   public void close() throws IOException {
+    indexWriterRAMManager.removeWriter();
     if (config.getCommitOnClose()) {
       shutdown();
     } else {
@@ -2445,6 +2451,7 @@ public void rollback() throws IOException {
     // Ensure that only one thread actually gets to do the
     // closing, and make sure no commit is also in progress:
     if (shouldClose(true)) {
+      indexWriterRAMManager.removeWriter();
       rollbackInternal();
     }
   }
@@ -6012,6 +6019,7 @@ private long maybeProcessEvents(long seqNo) throws IOException {
       seqNo = -seqNo;
       processEvents(true);
     }
+    indexWriterRAMManager.flushIfNecessary(config.flushPolicy);
     return seqNo;
   }
 

lucene/core/src/java/org/apache/lucene/index/IndexWriterConfig.java

@@ -142,7 +142,21 @@ public IndexWriterConfig() {
    * problem you should switch to {@link LogByteSizeMergePolicy} or {@link LogDocMergePolicy}.
    */
   public IndexWriterConfig(Analyzer analyzer) {
-    super(analyzer);
+    this(analyzer, new IndexWriterRAMManager(IndexWriterConfig.DEFAULT_RAM_BUFFER_SIZE_MB));
+  }
+
+  /**
+   * Creates a new config with the provided {@link IndexWriterRAMManager}. If you want to share a
+   * buffer between multiple {@link IndexWriter}, you will need to use this constructor as {@link
+   * IndexWriterConfig} maintains a 1:1 relationship with {@link IndexWriter}
+   */
+  public IndexWriterConfig(IndexWriterRAMManager indexWriterRAMManager) {
+    this(new StandardAnalyzer(), indexWriterRAMManager);
+  }
+
+  /** Creates a new config with the provided {@link Analyzer} and {@link IndexWriterRAMManager} */
+  public IndexWriterConfig(Analyzer analyzer, IndexWriterRAMManager indexWriterRAMManager) {
+    super(analyzer, indexWriterRAMManager);
   }
 
   /**
@@ -393,6 +407,11 @@ public double getRAMBufferSizeMB() {
     return super.getRAMBufferSizeMB();
   }
 
+  @Override
+  public IndexWriterRAMManager getIndexWriterRAMManager() {
+    return super.getIndexWriterRAMManager();
+  }
+
   /**
    * Information about merges, deletes and a message when maxFieldLength is reached will be printed
    * to this. Must not be null, but {@link InfoStream#NO_OUTPUT} may be used to suppress output.