Index: src/test/org/apache/lucene/index/TestNRT.java =================================================================== --- src/test/org/apache/lucene/index/TestNRT.java (revision 0) +++ src/test/org/apache/lucene/index/TestNRT.java (revision 0) @@ -0,0 +1,59 @@ +package org.apache.lucene.index; + +import java.util.Random; + +import org.apache.lucene.analysis.WhitespaceAnalyzer; +import org.apache.lucene.store.Directory; +import org.apache.lucene.store.MockRAMDirectory; +import org.apache.lucene.store.RAMDirectory; +import org.apache.lucene.util.LuceneTestCase; + +public class TestNRT extends LuceneTestCase { + Random random = new Random(); + + public void testSimple() throws Exception { + Directory primaryDir = new MockRAMDirectory(); + RAMDirectory ramDir = new MockRAMDirectory(); + + IndexWriter ramWriter = new IndexWriter(ramDir, new WhitespaceAnalyzer(), true, + IndexWriter.MaxFieldLength.LIMITED); + IndexWriter primaryWriter = new IndexWriter(primaryDir, new WhitespaceAnalyzer(), true, + IndexWriter.MaxFieldLength.LIMITED); + primaryWriter.setInfoStream(System.out); + NRT nrt = new NRT(ramWriter, primaryWriter, 1024*600); + + for (int i = 0; i < 100; i++) { + primaryWriter.addDocument(TestIndexWriterReader.createDocument(i, + "primary", 4)); + } + for (int i = 0; i < 100; i++) { + nrt.addDocument(TestIndexWriterReader.createDocument(i, "ram", 4)); + } + + IndexReader reader = nrt.getReader(); + assertEquals(200, reader.maxDoc()); + IndexReader ramReader = ramWriter.getReader(); + assertEquals(100, ramReader.maxDoc()); + IndexReader primaryReader = primaryWriter.getReader(); + assertEquals(100, primaryReader.maxDoc()); + + ramReader.close(); + primaryReader.close(); + reader.close(); + + nrt.flush(); + primaryReader = primaryWriter.getReader(); + assertEquals(200, primaryReader.maxDoc()); + ramReader = ramWriter.getReader(); + assertEquals(0, ramReader.maxDoc()); + + ramReader.close(); + primaryReader.close(); + + primaryWriter.close(); + ramWriter.close(); + + ramDir.close(); + primaryDir.close(); + } +} Index: src/java/org/apache/lucene/index/MergePolicy.java =================================================================== --- src/java/org/apache/lucene/index/MergePolicy.java (revision 832846) +++ src/java/org/apache/lucene/index/MergePolicy.java (working copy) @@ -85,10 +85,12 @@ final boolean useCompoundFile; boolean aborted; Throwable error; + IndexWriter writer; - public OneMerge(SegmentInfos segments, boolean useCompoundFile) { + public OneMerge(IndexWriter writer, SegmentInfos segments, boolean useCompoundFile) { if (0 == segments.size()) throw new RuntimeException("segments must include at least one segment"); + this.writer = writer; this.segments = segments; this.useCompoundFile = useCompoundFile; } Index: src/java/org/apache/lucene/index/LogMergePolicy.java =================================================================== --- src/java/org/apache/lucene/index/LogMergePolicy.java (revision 832846) +++ src/java/org/apache/lucene/index/LogMergePolicy.java (working copy) @@ -242,7 +242,7 @@ // First, enroll all "full" merges (size // mergeFactor) to potentially be run concurrently: while (last - maxNumSegments + 1 >= mergeFactor) { - spec.add(new OneMerge(infos.range(last-mergeFactor, last), useCompoundFile)); + spec.add(new OneMerge(writer, infos.range(last-mergeFactor, last), useCompoundFile)); last -= mergeFactor; } @@ -254,7 +254,7 @@ // Since we must optimize down to 1 segment, the // choice is simple: if (last > 1 || !isOptimized(infos.info(0))) - spec.add(new OneMerge(infos.range(0, last), useCompoundFile)); + spec.add(new OneMerge(writer, infos.range(0, last), useCompoundFile)); } else if (last > maxNumSegments) { // Take care to pick a partial merge that is @@ -282,7 +282,7 @@ } } - spec.add(new OneMerge(infos.range(bestStart, bestStart+finalMergeSize), useCompoundFile)); + spec.add(new OneMerge(writer, infos.range(bestStart, bestStart+finalMergeSize), useCompoundFile)); } } @@ -322,7 +322,7 @@ // deletions, so force a merge now: if (verbose()) message(" add merge " + firstSegmentWithDeletions + " to " + (i-1) + " inclusive"); - spec.add(new OneMerge(segmentInfos.range(firstSegmentWithDeletions, i), useCompoundFile)); + spec.add(new OneMerge(writer, segmentInfos.range(firstSegmentWithDeletions, i), useCompoundFile)); firstSegmentWithDeletions = i; } } else if (firstSegmentWithDeletions != -1) { @@ -331,7 +331,7 @@ // mergeFactor segments if (verbose()) message(" add merge " + firstSegmentWithDeletions + " to " + (i-1) + " inclusive"); - spec.add(new OneMerge(segmentInfos.range(firstSegmentWithDeletions, i), useCompoundFile)); + spec.add(new OneMerge(writer, segmentInfos.range(firstSegmentWithDeletions, i), useCompoundFile)); firstSegmentWithDeletions = -1; } } @@ -339,7 +339,7 @@ if (firstSegmentWithDeletions != -1) { if (verbose()) message(" add merge " + firstSegmentWithDeletions + " to " + (numSegments-1) + " inclusive"); - spec.add(new OneMerge(segmentInfos.range(firstSegmentWithDeletions, numSegments), useCompoundFile)); + spec.add(new OneMerge(writer, segmentInfos.range(firstSegmentWithDeletions, numSegments), useCompoundFile)); } return spec; @@ -439,7 +439,7 @@ spec = new MergeSpecification(); if (verbose()) message(" " + start + " to " + end + ": add this merge"); - spec.add(new OneMerge(infos.range(start, end), useCompoundFile)); + spec.add(new OneMerge(writer, infos.range(start, end), useCompoundFile)); } else if (verbose()) message(" " + start + " to " + end + ": contains segment over maxMergeSize or maxMergeDocs; skipping"); Index: src/java/org/apache/lucene/index/NRT.java =================================================================== --- src/java/org/apache/lucene/index/NRT.java (revision 0) +++ src/java/org/apache/lucene/index/NRT.java (revision 0) @@ -0,0 +1,121 @@ +package org.apache.lucene.index; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +import org.apache.lucene.analysis.Analyzer; +import org.apache.lucene.document.Document; +import org.apache.lucene.search.Query; +import org.apache.lucene.store.RAMDirectory; + +public class NRT { + private IndexWriter ramWriter; + + private IndexWriter primaryWriter; + + RAMDirectory ramDirectory; + + long maxRam; + + public NRT(IndexWriter ramWriter, IndexWriter primaryWriter, long maxRam) { + this.ramWriter = ramWriter; + ramDirectory = (RAMDirectory) ramWriter.getDirectory(); + this.primaryWriter = primaryWriter; + this.maxRam = maxRam; + } + + /** + * Flush the RAM segments and buffer to the primary writer. + * + * @throws IOException + */ + public synchronized void flush() throws IOException { + ramWriter.flush(false, true, true); + SegmentInfos ramInfos = ramWriter.getSegmentInfosCopy(); + MergePolicy.OneMerge merge = new MergePolicy.OneMerge(ramWriter, ramInfos, + primaryWriter.getUseCompoundFile()); + primaryWriter.mergeIn(merge); + ramWriter.removeInfos(ramInfos); + } + + public IndexReader getReader() throws IOException { + IndexReader[] ramReaders = ramWriter.getReader().getSequentialSubReaders(); + IndexReader[] primaryReaders = primaryWriter.getReader() + .getSequentialSubReaders(); + List readers = new ArrayList(ramReaders.length + + primaryReaders.length); + int i = 0; + Set infos = new HashSet(); + for (int x = 0; x < primaryReaders.length; x++) { + SegmentInfo info = ((SegmentReader) primaryReaders[x]).getSegmentInfo(); + infos.add(info); + readers.add(primaryReaders[x]); + } + for (int x = 0; x < ramReaders.length; x++) { + SegmentInfo info = ((SegmentReader) ramReaders[x]).getSegmentInfo(); + if (!infos.contains(info)) { + infos.add(info); + readers.add(ramReaders[x]); + } + } + MultiReader reader = new MultiReader((IndexReader[]) readers + .toArray(new IndexReader[0])); + return reader; + } + + public long getRAMSize() { + return ramWriter.ramSizeInBytes() + ramDirectory.sizeInBytes(); + } + + public void addDocument(Document doc) throws IOException, + InterruptedException { + addDocument(doc, primaryWriter.getAnalyzer()); + } + + public void addDocument(Document doc, Analyzer analyzer) throws IOException, + InterruptedException { + ramWriter.addDocument(doc, analyzer); + ifFlush(); + } + + public void deleteDocuments(Query query) throws CorruptIndexException, + IOException, InterruptedException { + ramWriter.deleteDocuments(query); + primaryWriter.deleteDocuments(query); + ifFlush(); + } + + public void deleteDocuments(Query... queries) throws CorruptIndexException, + IOException, InterruptedException { + ramWriter.deleteDocuments(queries); + primaryWriter.deleteDocuments(queries); + ifFlush(); + } + + public void updateDocument(Term term, Document doc, Analyzer analyzer) + throws CorruptIndexException, IOException, InterruptedException { + ramWriter.updateDocument(term, doc, analyzer); + primaryWriter.deleteDocuments(term); + ifFlush(); + } + + public void updateDocument(Term term, Document doc) + throws CorruptIndexException, IOException, InterruptedException { + ramWriter.updateDocument(term, doc); + primaryWriter.deleteDocuments(term); + ifFlush(); + } + + private void ifFlush() throws IOException, InterruptedException { + if (getRAMSize() > maxRam) { + long ramSizeBytes = ramWriter.ramSizeInBytes(); + long ramDirBytes = ramDirectory.sizeInBytes(); + // if (infoStream != null) + // infoStream.println("ramSize:"+ramSizeBytes+" ramDir:"+ramDirBytes); + flush(); + } + } +} Index: src/java/org/apache/lucene/index/IndexWriter.java =================================================================== --- src/java/org/apache/lucene/index/IndexWriter.java (revision 832846) +++ src/java/org/apache/lucene/index/IndexWriter.java (working copy) @@ -3060,7 +3060,28 @@ private boolean hasExternalSegments() { return segmentInfos.hasExternalSegments(directory); } - + + final void mergeIn(MergePolicy.OneMerge merge) throws CorruptIndexException, IOException { + synchronized (this) { + for (SegmentInfo info : merge.segments) { + segmentInfos.add(info); + } + if (registerMerge(merge)) { + pendingMerges.remove(merge); + runningMerges.add(merge); + } + } + merge(merge); + } + + synchronized SegmentInfos getSegmentInfosCopy() { + SegmentInfos copy = new SegmentInfos(); + for (SegmentInfo info : segmentInfos) { + copy.add(info); + } + return copy; + } + /* If any of our segments are using a directory != ours * then we have to either copy them over one by one, merge * them (if merge policy has chosen to) or wait until @@ -3089,7 +3110,7 @@ info = segmentInfos.info(i); if (info.dir != directory) { done = false; - final MergePolicy.OneMerge newMerge = new MergePolicy.OneMerge(segmentInfos.range(i, 1+i), mergePolicy instanceof LogMergePolicy && getUseCompoundFile()); + final MergePolicy.OneMerge newMerge = new MergePolicy.OneMerge(this, segmentInfos.range(i, 1+i), mergePolicy instanceof LogMergePolicy && getUseCompoundFile()); // Returns true if no running merge conflicts // with this one (and, records this merge as @@ -3827,7 +3848,7 @@ // If the merged segments had pending changes, clear // them so that they don't bother writing them to // disk, updating SegmentInfo, etc.: - readerPool.clear(merge.segments); + merge.writer.readerPool.clear(merge.segments); if (merge.optimize) segmentsToOptimize.add(merge.info); @@ -4211,7 +4232,7 @@ // Hold onto the "live" reader; we will use this to // commit merged deletes - SegmentReader reader = merge.readers[i] = readerPool.get(info, merge.mergeDocStores, + SegmentReader reader = merge.readers[i] = merge.writer.readerPool.get(info, merge.mergeDocStores, MERGE_READ_BUFFER_SIZE, -1); @@ -4292,7 +4313,7 @@ for (int i=0;i