Index: CHANGES.txt =================================================================== --- CHANGES.txt (revision 772355) +++ CHANGES.txt (working copy) @@ -299,6 +299,12 @@ cross-correlate Spans from different fields. (Paul Cowan and Chris Hostetter) +25. LUCENE-1313: Near realtime search enhancement. IndexWriter + accepts a RAMDirectory that segments are flushed to (instead of + to the real directory). This can be faster than flushing and + merging small segments using only an FSDirectory. + (Jason Rutherglen via Mike McCandless) + Optimizations 1. LUCENE-1427: Fixed QueryWrapperFilter to not waste time computing Index: src/java/org/apache/lucene/index/ConcurrentMergeScheduler.java =================================================================== --- src/java/org/apache/lucene/index/ConcurrentMergeScheduler.java (revision 773472) +++ src/java/org/apache/lucene/index/ConcurrentMergeScheduler.java (working copy) @@ -207,7 +207,7 @@ } if (verbose()) - message(" consider merge " + merge.segString(dir)); + message(" consider merge " + merge.segString()); assert mergeThreadCount() < maxThreadCount; @@ -296,7 +296,7 @@ if (merge != null) { writer.mergeInit(merge); if (verbose()) - message(" merge thread: do another merge " + merge.segString(dir)); + message(" merge thread: do another merge " + merge.segString()); } else break; } @@ -328,7 +328,7 @@ MergePolicy.OneMerge merge = getRunningMerge(); if (merge == null) merge = startMerge; - return "merge thread: " + merge.segString(dir); + return "merge thread: " + merge.segString(); } } Index: src/java/org/apache/lucene/index/DocFieldProcessorPerThread.java =================================================================== --- src/java/org/apache/lucene/index/DocFieldProcessorPerThread.java (revision 773472) +++ src/java/org/apache/lucene/index/DocFieldProcessorPerThread.java (working copy) @@ -23,6 +23,7 @@ import java.io.IOException; import org.apache.lucene.document.Document; import org.apache.lucene.document.Fieldable; +import org.apache.lucene.store.Directory; /** * Gathers all Fieldables for a document under the same @@ -146,7 +147,6 @@ } public DocumentsWriter.DocWriter processDocument() throws IOException { - consumer.startDocument(); final Document doc = docState.doc; Index: src/java/org/apache/lucene/index/DocumentsWriter.java =================================================================== --- src/java/org/apache/lucene/index/DocumentsWriter.java (revision 773472) +++ src/java/org/apache/lucene/index/DocumentsWriter.java (working copy) @@ -25,6 +25,7 @@ import org.apache.lucene.search.Scorer; import org.apache.lucene.search.Weight; import org.apache.lucene.store.Directory; +import org.apache.lucene.store.RAMDirectory; import org.apache.lucene.store.AlreadyClosedException; import org.apache.lucene.util.ArrayUtil; @@ -110,7 +111,6 @@ final class DocumentsWriter { IndexWriter writer; - Directory directory; String segment; // Current segment we are working on private String docStoreSegment; // Current doc-store segment we are writing @@ -131,14 +131,15 @@ boolean flushPending; // True when a thread has decided to flush boolean bufferIsFull; // True when it's time to write segment private boolean aborting; // True if an abort is pending + Directory flushToDir; private DocFieldProcessor docFieldProcessor; PrintStream infoStream; int maxFieldLength = IndexWriter.DEFAULT_MAX_FIELD_LENGTH; Similarity similarity; - List newFiles; + private long r; static class DocState { DocumentsWriter docWriter; @@ -235,6 +236,9 @@ private long ramBufferSize = (long) (IndexWriter.DEFAULT_RAM_BUFFER_SIZE_MB*1024*1024); private long waitQueuePauseBytes = (long) (ramBufferSize*0.1); private long waitQueueResumeBytes = (long) (ramBufferSize*0.05); + + private long ramDirSizeMax = 0; + private long ramDirSize = 0; // If we've allocated 5% over our RAM budget, we then // free down to 95% @@ -256,11 +260,10 @@ synchronized void setFlushedDocCount(int n) { flushedDocCount = n; } - + private boolean closed; - DocumentsWriter(Directory directory, IndexWriter writer, IndexingChain indexingChain) throws IOException { - this.directory = directory; + DocumentsWriter(IndexWriter writer, IndexingChain indexingChain) throws IOException { this.writer = writer; this.similarity = writer.getSimilarity(); flushedDocCount = writer.maxDoc(); @@ -270,7 +273,7 @@ docFieldProcessor = (DocFieldProcessor) consumer; } } - + /** Returns true if any of the fields in the current * buffered docs have omitTermFreqAndPositions==false */ boolean hasProx() { @@ -285,7 +288,15 @@ for(int i=0;i= ramBufferSize || numBytesAlloc >= freeTrigger); } - + /** Does the synchronized work to finish/flush the * inverted document. */ private void finishDocument(DocumentsWriterThreadState perThread, DocWriter docWriter) throws IOException { @@ -1111,10 +1156,9 @@ // TODO FI: this is not flexible -- we can't hardwire // extensions in here: - private long segmentSize(String segmentName) throws IOException { + private long segmentSize(String segmentName, Directory directory) throws IOException { // Used only when infoStream != null assert infoStream != null; - long size = directory.fileLength(segmentName + ".tii") + directory.fileLength(segmentName + ".tis") + directory.fileLength(segmentName + ".frq") + @@ -1123,10 +1167,15 @@ final String normFileName = segmentName + ".nrm"; if (directory.fileExists(normFileName)) size += directory.fileLength(normFileName); - + return size; } - + + public long fileLength(String name) throws IOException { + Directory dir = writer.getInternalDirectory(); + return dir.fileLength(name); + } + // Coarse estimates used to measure RAM usage of buffered deletes final static int OBJECT_HEADER_BYTES = 8; final static int POINTER_NUM_BYTE = 4; @@ -1358,7 +1407,7 @@ } } } - + final WaitQueue waitQueue = new WaitQueue(); private class WaitQueue { Index: src/java/org/apache/lucene/index/IndexWriter.java =================================================================== --- src/java/org/apache/lucene/index/IndexWriter.java (revision 773472) +++ src/java/org/apache/lucene/index/IndexWriter.java (working copy) @@ -28,6 +28,8 @@ import org.apache.lucene.store.LockObtainFailedException; import org.apache.lucene.store.AlreadyClosedException; import org.apache.lucene.store.BufferedIndexInput; +import org.apache.lucene.store.RAMDirectory; +import org.apache.lucene.store.FileSwitchDirectory; import org.apache.lucene.util.Constants; import java.io.File; @@ -286,6 +288,18 @@ * set (see {@link #setInfoStream}). */ public final static int MAX_TERM_LENGTH = DocumentsWriter.MAX_TERM_LENGTH; + + /** + * File extensions for the FileSwitchDirectory used when NRT is on + */ + public final static Set SWITCH_FILE_EXTS = new HashSet(); + static { + SWITCH_FILE_EXTS.add("fdx"); + SWITCH_FILE_EXTS.add("fdt"); + SWITCH_FILE_EXTS.add("tvx"); + SWITCH_FILE_EXTS.add("tvf"); + SWITCH_FILE_EXTS.add("tvd"); + } /** * Default for {@link #getMaxSyncPauseSeconds}. On @@ -335,10 +349,10 @@ private SegmentInfos segmentInfos = new SegmentInfos(); // the segments - private DocumentsWriter docWriter; + DocumentsWriter docWriter; private IndexFileDeleter deleter; - private Set segmentsToOptimize = new HashSet(); // used by optimize to note those needing optimization + Set segmentsToOptimize = new HashSet(); // used by optimize to note those needing optimization private Lock writeLock; @@ -352,8 +366,10 @@ // merges private HashSet mergingSegments = new HashSet(); - private MergePolicy mergePolicy = new LogByteSizeMergePolicy(); - private MergeScheduler mergeScheduler = new ConcurrentMergeScheduler(); + MergePolicy mergePolicy = new LogByteSizeMergePolicy(); + RAMMergePolicy ramMergePolicy = new RAMMergePolicy(); + private NRTMergePolicy nrtMergePolicy; + MergeScheduler mergeScheduler = new ConcurrentMergeScheduler(); private LinkedList pendingMerges = new LinkedList(); private Set runningMerges = new HashSet(); private List mergeExceptions = new ArrayList(); @@ -370,6 +386,7 @@ private Thread writeThread; // non-null if any thread holds write lock final ReaderPool readerPool = new ReaderPool(); private int upgradeCount; + private FileSwitchDirectory switchDirectory; // This is a "write once" variable (like the organic dye // on a DVD-R that may or may not be heated by a laser and @@ -382,6 +399,8 @@ // readers. private volatile boolean poolReaders; + boolean ramNrt; + /** * Expert: returns a readonly reader containing all * current updates. Flush is called automatically. This @@ -441,7 +460,9 @@ // this method is called: poolReaders = true; - flush(true, true, true); + // when getReader is called we assume we want to try to flush to ram + // (we may not be able to flush to ram as there may not be enough available ram) + flush(true, true, true, true); // Prevent segmentInfos from changing while opening the // reader; in theory we could do similar retry logic, @@ -450,7 +471,31 @@ return new ReadOnlyMultiSegmentReader(this, segmentInfos); } } - + + /** + * Ram directory (switchDirectory) segments reader + */ + // for testing + IndexReader getRamReader() throws IOException { + flush(true, true, true, true); + + synchronized(this) { + return new ReadOnlyMultiSegmentReader(this, getRamDirSegmentInfos()); + } + } + + /** + * Primary directory segments reader + */ + //for testing + IndexReader getPrimaryReader() throws IOException { + flush(true, true, true, true); + + synchronized(this) { + return new ReadOnlyMultiSegmentReader(this, getInfosByDir(segmentInfos, directory)); + } + } + /** Holds shared SegmentReader instances. IndexWriter uses * SegmentReaders for 1) applying deletes, 2) doing * merges, 3) handing out a real-time reader. This pool @@ -680,7 +725,16 @@ assert readCount >= 0; notifyAll(); } - + + boolean isRamMerge(MergePolicy.OneMerge merge) { + if (ramNrt) { + if (merge.directory == switchDirectory) return true; + return false; + } else { + return false; + } + } + synchronized final boolean isOpen(boolean includePendingClose) { return !(closed || (includePendingClose && closing)); } @@ -725,11 +779,16 @@ * an exception if the mergePolicy is not a LogMergePolicy. */ private LogMergePolicy getLogMergePolicy() { - if (mergePolicy instanceof LogMergePolicy) - return (LogMergePolicy) mergePolicy; + MergePolicy mp = mergePolicy; + if (mp instanceof LogMergePolicy) + return (LogMergePolicy) mp; else throw new IllegalArgumentException("this method can only be called when the merge policy is the default LogMergePolicy"); } + + public MergePolicy getRAMMergePolicy() { + return ramMergePolicy; + } /**

Get the current setting of whether newly flushed * segments will use the compound file format. Note that @@ -849,11 +908,43 @@ throws CorruptIndexException, LockObtainFailedException, IOException { init(FSDirectory.getDirectory(path), a, create, true, null, false, mfl.getLimit(), null, null); } - + /** * Constructs an IndexWriter for the index in path. * Text will be analyzed with a. If create * is true, then a new, empty index will be created in + * path, replacing the index already there, + * if any. + * + *

NOTE: autoCommit (see above) is set to false with this + * constructor. + * + * @param path the path to the index directory + * @param a the analyzer to use + * @param create true to create the index or overwrite + * the existing one; false to append to the existing + * index + * @param mfl Maximum field length in number of tokens/terms: LIMITED, UNLIMITED, or user-specified + * via the MaxFieldLength constructor. + * @throws CorruptIndexException if the index is corrupt + * @throws LockObtainFailedException if another writer + * has this index open (write.lock could not + * be obtained) + * @throws IOException if the directory cannot be read/written to, or + * if it does not exist and create is + * false or if there is any other low-level + * IO error + */ + public IndexWriter(String path, Analyzer a, boolean create, MaxFieldLength mfl, boolean flushToRam) + throws CorruptIndexException, LockObtainFailedException, IOException { + init(FSDirectory.getDirectory(path), a, create, true, null, false, mfl.getLimit(), null, null); + } + + /** + * Constructs an IndexWriter for the index in path. + * Text will be analyzed with a. If create + * is true, then a new, empty index will be created in * path, replacing the index already there, if any. * * @param path the path to the index directory @@ -969,18 +1060,25 @@ throws CorruptIndexException, LockObtainFailedException, IOException { init(d, a, create, false, null, false, mfl.getLimit(), null, null); } - + /** * Constructs an IndexWriter for the index in d. * Text will be analyzed with a. If create * is true, then a new, empty index will be created in * d, replacing the index already there, if any. * + *

NOTE: autoCommit (see above) is set to false with this + * constructor. + * * @param d the index directory * @param a the analyzer to use * @param create true to create the index or overwrite * the existing one; false to append to the existing * index + * @param mfl Maximum field length in number of terms/tokens: LIMITED, UNLIMITED, or user-specified + * via the MaxFieldLength constructor. + * @param flushToRam If many segments are being created opened for reading (@see #getReader) set this to true * @throws CorruptIndexException if the index is corrupt * @throws LockObtainFailedException if another writer * has this index open (write.lock could not @@ -989,6 +1087,31 @@ * if it does not exist and create is * false or if there is any other low-level * IO error + */ + public IndexWriter(Directory d, Analyzer a, boolean create, MaxFieldLength mfl, boolean flushToRam) + throws CorruptIndexException, LockObtainFailedException, IOException { + init(d, a, create, false, null, false, mfl.getLimit(), null, null, flushToRam, null); + } + + /** + * Constructs an IndexWriter for the index in d. + * Text will be analyzed with a. If create + * is true, then a new, empty index will be created in + * d, replacing the index already there, if any. + * + * @param d the index directory + * @param a the analyzer to use + * @param create true to create the index or overwrite + * the existing one; false to append to the existing + * index + * @throws CorruptIndexException if the index is corrupt + * @throws LockObtainFailedException if another writer + * has this index open (write.lock could not + * be obtained) + * @throws IOException if the directory cannot be read/written to, or + * if it does not exist and create is + * false or if there is any other low-level + * IO error * @deprecated This constructor will be removed in the 3.0 * release, and call {@link #commit()} when needed. * Use {@link #IndexWriter(Directory,Analyzer,boolean,MaxFieldLength)} instead. @@ -1099,7 +1222,7 @@ throws CorruptIndexException, LockObtainFailedException, IOException { init(FSDirectory.getDirectory(path), a, true, null, true, DEFAULT_MAX_FIELD_LENGTH, null, null); } - + /** * Constructs an IndexWriter for the index in * d, first creating it if it does not @@ -1126,15 +1249,22 @@ throws CorruptIndexException, LockObtainFailedException, IOException { init(d, a, false, null, false, mfl.getLimit(), null, null); } - + /** * Constructs an IndexWriter for the index in * d, first creating it if it does not * already exist. Text will be analyzed with * a. * + *

NOTE: autoCommit (see above) is set to false with this + * constructor. + * * @param d the index directory * @param a the analyzer to use + * @param mfl Maximum field length in number of terms/tokens: LIMITED, UNLIMITED, or user-specified + * via the MaxFieldLength constructor. + * @param flushToRam If many segments are being created opened for reading (@see #getReader) set this to true * @throws CorruptIndexException if the index is corrupt * @throws LockObtainFailedException if another writer * has this index open (write.lock could not @@ -1142,6 +1272,32 @@ * @throws IOException if the directory cannot be * read/written to or if there is any other low-level * IO error + */ + public IndexWriter(Directory d, Analyzer a, MaxFieldLength mfl, boolean flushToRam) + throws CorruptIndexException, LockObtainFailedException, IOException { + init(d, a, false, null, false, mfl.getLimit(), null, null, flushToRam); + } + + IndexWriter(Directory d, Analyzer a, MaxFieldLength mfl, boolean flushToRam, Directory ramDir) + throws CorruptIndexException, LockObtainFailedException, IOException { + init(d, a, false, null, false, mfl.getLimit(), null, null, flushToRam, ramDir); + } + + /** + * Constructs an IndexWriter for the index in + * d, first creating it if it does not + * already exist. Text will be analyzed with + * a. + * + * @param d the index directory + * @param a the analyzer to use + * @throws CorruptIndexException if the index is corrupt + * @throws LockObtainFailedException if another writer + * has this index open (write.lock could not + * be obtained) + * @throws IOException if the directory cannot be + * read/written to or if there is any other low-level + * IO error * @deprecated This constructor will be removed in the 3.0 release. * Use {@link * #IndexWriter(Directory,Analyzer,MaxFieldLength)} @@ -1261,7 +1417,7 @@ throws CorruptIndexException, LockObtainFailedException, IOException { init(d, a, false, deletionPolicy, autoCommit, DEFAULT_MAX_FIELD_LENGTH, null, null); } - + /** * Expert: constructs an IndexWriter with a custom {@link * IndexDeletionPolicy}, for the index in d. @@ -1294,7 +1450,7 @@ throws CorruptIndexException, LockObtainFailedException, IOException { init(d, a, create, false, deletionPolicy, false, mfl.getLimit(), null, null); } - + /** * Expert: constructs an IndexWriter with a custom {@link * IndexDeletionPolicy} and {@link IndexingChain}, @@ -1331,7 +1487,7 @@ throws CorruptIndexException, LockObtainFailedException, IOException { init(d, a, create, false, deletionPolicy, false, mfl.getLimit(), indexingChain, commit); } - + /** * Expert: constructs an IndexWriter with a custom {@link * IndexDeletionPolicy}, for the index in d. @@ -1405,27 +1561,52 @@ throws CorruptIndexException, LockObtainFailedException, IOException { init(d, a, false, false, deletionPolicy, false, mfl.getLimit(), null, commit); } - + private void init(Directory d, Analyzer a, boolean closeDir, IndexDeletionPolicy deletionPolicy, - boolean autoCommit, int maxFieldLength, IndexingChain indexingChain, IndexCommit commit) + boolean autoCommit, int maxFieldLength, IndexingChain indexingChain, IndexCommit commit) throws CorruptIndexException, LockObtainFailedException, IOException { + init(d, a, closeDir, deletionPolicy, autoCommit, maxFieldLength, indexingChain, commit, false); + } + + private void init(Directory d, Analyzer a, boolean closeDir, IndexDeletionPolicy deletionPolicy, + boolean autoCommit, int maxFieldLength, IndexingChain indexingChain, IndexCommit commit, boolean flushToRam) + throws CorruptIndexException, LockObtainFailedException, IOException { + init(d, a, closeDir, deletionPolicy, autoCommit, maxFieldLength, indexingChain, commit, flushToRam, (Directory)null); + } + + private void init(Directory d, Analyzer a, boolean closeDir, IndexDeletionPolicy deletionPolicy, + boolean autoCommit, int maxFieldLength, IndexingChain indexingChain, IndexCommit commit, boolean flushToRam, Directory ramDir) + throws CorruptIndexException, LockObtainFailedException, IOException { if (IndexReader.indexExists(d)) { - init(d, a, false, closeDir, deletionPolicy, autoCommit, maxFieldLength, indexingChain, commit); + init(d, a, false, closeDir, deletionPolicy, autoCommit, maxFieldLength, indexingChain, commit, flushToRam, ramDir); } else { - init(d, a, true, closeDir, deletionPolicy, autoCommit, maxFieldLength, indexingChain, commit); + init(d, a, true, closeDir, deletionPolicy, autoCommit, maxFieldLength, indexingChain, commit, flushToRam, ramDir); } } - + private void init(Directory d, Analyzer a, final boolean create, boolean closeDir, + IndexDeletionPolicy deletionPolicy, boolean autoCommit, int maxFieldLength, + IndexingChain indexingChain, IndexCommit commit) throws CorruptIndexException, LockObtainFailedException, IOException { + init(d, a, create, closeDir, deletionPolicy, autoCommit, maxFieldLength, indexingChain, commit, false, (Directory)null); + } + + private void init(Directory d, Analyzer a, final boolean create, boolean closeDir, IndexDeletionPolicy deletionPolicy, boolean autoCommit, int maxFieldLength, - IndexingChain indexingChain, IndexCommit commit) + IndexingChain indexingChain, IndexCommit commit, boolean ramNrt, Directory ramDir) throws CorruptIndexException, LockObtainFailedException, IOException { this.closeDir = closeDir; directory = d; + this.ramNrt = ramNrt; + if (ramDir == null) ramDir = new RAMDirectory(); + this.switchDirectory = new FileSwitchDirectory(SWITCH_FILE_EXTS, directory, ramDir, false); analyzer = a; setMessageID(defaultInfoStream); this.maxFieldLength = maxFieldLength; + if (ramNrt) { + nrtMergePolicy = new NRTMergePolicy(this); + } + if (indexingChain == null) indexingChain = DocumentsWriter.DefaultIndexingChain; @@ -1493,13 +1674,15 @@ this.autoCommit = autoCommit; setRollbackSegmentInfos(segmentInfos); - docWriter = new DocumentsWriter(directory, this, indexingChain); + docWriter = new DocumentsWriter(this, indexingChain); docWriter.setInfoStream(infoStream); docWriter.setMaxFieldLength(maxFieldLength); - + if (ramNrt) { + setRAMBufferSizeMB(IndexWriter.DEFAULT_RAM_BUFFER_SIZE_MB); // initialize ram dir size + } // Default deleter (for backwards compatibility) is // KeepOnlyLastCommitDeleter: - deleter = new IndexFileDeleter(directory, + deleter = new IndexFileDeleter(switchDirectory, deletionPolicy == null ? new KeepOnlyLastCommitDeletionPolicy() : deletionPolicy, segmentInfos, infoStream, docWriter); @@ -1523,20 +1706,61 @@ throw e; } } - + + Directory getInternalDirectory() { + if (ramNrt) + return switchDirectory; + else return directory; + } + + long getRamDirSize() { + return ((RAMDirectory)switchDirectory.getSecondaryDir()).sizeInBytes(); + } + + static long size(Directory dir) throws IOException { + String[] files = dir.listAll(); + long c = 0; + for (int x=0; x < files.length; x++) { + c +=dir.fileLength(files[x]); + } + return c; + } + + public Directory getRAMDirectory() { + return switchDirectory.getSecondaryDir(); + } + private synchronized void setRollbackSegmentInfos(SegmentInfos infos) { rollbackSegmentInfos = (SegmentInfos) infos.clone(); - assert !rollbackSegmentInfos.hasExternalSegments(directory); + assert !rollbackSegmentInfos.hasExternalSegments(getInternalDirectory()); rollbackSegments = new HashMap(); final int size = rollbackSegmentInfos.size(); for(int i=0;iDetermines the largest segment (measured by * document count) that may be merged with other segments. * Small values (e.g., less than 10,000) are best for @@ -1746,11 +1969,20 @@ if (mb == DISABLE_AUTO_FLUSH && getMaxBufferedDocs() == DISABLE_AUTO_FLUSH) throw new IllegalArgumentException( "at least one of ramBufferSize and maxBufferedDocs must be enabled"); - docWriter.setRAMBufferSizeMB(mb); + + if (ramNrt) { + // if NRT is on, then set the ramBufferSize to half + // set the ramDirSize to the other half + double ramMB = mb * 0.5; + docWriter.setRAMBufferSizeMB(ramMB); + docWriter.setRAMDirSizeMB(ramMB); + } else { + docWriter.setRAMBufferSizeMB(mb); + } if (infoStream != null) message("setRAMBufferSizeMB " + mb); } - + /** * Returns the value set by {@link #setRAMBufferSizeMB} if enabled. */ @@ -2064,9 +2296,10 @@ if (waitForMerges) // Give merge scheduler last chance to run, in case // any pending merges are waiting: - mergeScheduler.merge(this); + scheduleMerge(); mergePolicy.close(); + if (ramMergePolicy != null) ramMergePolicy.close(); finishMerges(waitForMerges); @@ -2088,9 +2321,12 @@ deleter.close(); } - if (closeDir) + if (closeDir) { directory.close(); - + Directory ramDir = switchDirectory.getSecondaryDir(); + ramDir.close(); + } + if (writeLock != null) { writeLock.release(); // release write lock writeLock = null; @@ -2149,7 +2385,8 @@ final String compoundFileName = docStoreSegment + "." + IndexFileNames.COMPOUND_FILE_STORE_EXTENSION; try { - CompoundFileWriter cfsWriter = new CompoundFileWriter(directory, compoundFileName); + Directory dir = getDirectory(); // compound files are always written to the primary dir (not the FSD) + CompoundFileWriter cfsWriter = new CompoundFileWriter(dir, compoundFileName); final Iterator it = docWriter.closedFiles().iterator(); while(it.hasNext()) cfsWriter.addFile((String) it.next()); @@ -2531,7 +2768,24 @@ handleOOM(oom, "updateDocument"); } } - + + //for test purpose + final synchronized SegmentInfos getSegmentInfos() { + return segmentInfos; + } + + //for test purpose + final synchronized SegmentInfos getSegmentInfos(Directory dir) { + SegmentInfos infos = new SegmentInfos(); + for (int x=0; x < infos.size(); x++) { + SegmentInfo info = infos.info(x); + if (info.dir == dir) { + infos.add(info); + } + } + return infos; + } + // for test purpose final synchronized int getSegmentCount(){ return segmentInfos.size(); @@ -2690,7 +2944,7 @@ if (infoStream != null) message("optimize: index now " + segString()); - flush(true, false, true); + flush(true, false, true, true); // if there's room, flush the buffer to the ramdir synchronized(this) { resetMergeExceptions(); @@ -2706,6 +2960,7 @@ final MergePolicy.OneMerge merge = (MergePolicy.OneMerge) it.next(); merge.optimize = true; merge.maxNumSegmentsOptimize = maxNumSegments; + merge.directory = directory; } it = runningMerges.iterator(); @@ -2713,6 +2968,7 @@ final MergePolicy.OneMerge merge = (MergePolicy.OneMerge) it.next(); merge.optimize = true; merge.maxNumSegmentsOptimize = maxNumSegments; + merge.directory = directory; } } @@ -2733,7 +2989,7 @@ for(int i=0;i 0) { + segmentsToOptimize = new HashSet(); + final int numSegments = ramInfos.size(); + for(int i=0;i 0; @@ -2912,24 +3238,27 @@ final MergePolicy.MergeSpecification spec; if (optimize) { - spec = mergePolicy.findMergesForOptimize(segmentInfos, this, maxNumSegmentsOptimize, segmentsToOptimize); - + spec = getCurrentMergePolicy().findMergesForOptimize(segmentInfos, this, maxNumSegmentsOptimize, segmentsToOptimize); if (spec != null) { final int numMerges = spec.merges.size(); for(int i=0;iAfter this completes, the index is optimized.

*

The provided IndexReaders are not closed.

@@ -3873,7 +4296,6 @@ } private final void prepareCommit(String commitUserData, boolean internal) throws CorruptIndexException, IOException { - if (hitOOM) { throw new IllegalStateException("this writer hit an OutOfMemoryError; cannot commit"); } @@ -3886,9 +4308,10 @@ if (infoStream != null) message("prepareCommit: flush"); - - flush(true, true, true); - + + flush(true, true, true, false); + // merge ram segments to disk synchronously + resolveRamSegments(); startCommit(0, commitUserData); } @@ -3983,9 +4406,38 @@ if (infoStream != null) message("commit: done"); } - + + public static SegmentInfos getInfosByDir(SegmentInfos infos, Directory dir) throws IOException { + SegmentInfos dirInfos = new SegmentInfos(); + for (int x=0; x < infos.size(); x++) { + SegmentInfo info = infos.info(x); + if (info.dir == dir) { + dirInfos.add(info); + } + } + return dirInfos; + } + + synchronized SegmentInfos getRamDirSegmentInfos() throws IOException { + return getInfosByDir(segmentInfos, switchDirectory); + } + /** - * Flush all in-memory buffered udpates (adds and deletes) + * Flush all in-memory buffered updates (adds and deletes) + * to the Directory. By default we flush to the ram dir + * @param triggerMerge if true, we may merge segments (if + * deletes or docs were flushed) if necessary + * @param flushDocStores if false we are allowed to keep + * doc stores open to share with the next segment + * @param flushDeletes whether pending deletes should also + * be flushed + */ + protected final void flush(boolean triggerMerge, boolean flushDocStores, boolean flushDeletes) throws CorruptIndexException, IOException { + flush(triggerMerge, flushDocStores, flushDeletes, (ramNrt ? true : false)); + } + + /** + * Flush all in-memory buffered updates (adds and deletes) * to the Directory. * @param triggerMerge if true, we may merge segments (if * deletes or docs were flushed) if necessary @@ -3994,19 +4446,18 @@ * @param flushDeletes whether pending deletes should also * be flushed */ - protected final void flush(boolean triggerMerge, boolean flushDocStores, boolean flushDeletes) throws CorruptIndexException, IOException { - // We can be called during close, when closing==true, so we must pass false to ensureOpen: + protected final void flush(boolean triggerMerge, boolean flushDocStores, boolean flushDeletes, boolean flushToRam) throws CorruptIndexException, IOException { ensureOpen(false); - if (doFlush(flushDocStores, flushDeletes) && triggerMerge) + if (doFlush(flushDocStores, flushDeletes, flushToRam) && triggerMerge) maybeMerge(); } // TODO: this method should not have to be entirely // synchronized, ie, merges should be allowed to commit // even while a flush is happening - private synchronized final boolean doFlush(boolean flushDocStores, boolean flushDeletes) throws CorruptIndexException, IOException { + private synchronized final boolean doFlush(boolean flushDocStores, boolean flushDeletes, boolean flushToRam) throws CorruptIndexException, IOException { try { - return doFlushInternal(flushDocStores, flushDeletes); + return doFlushInternal(flushDocStores, flushDeletes, flushToRam); } finally { docWriter.clearFlushPending(); } @@ -4015,7 +4466,7 @@ // TODO: this method should not have to be entirely // synchronized, ie, merges should be allowed to commit // even while a flush is happening - private synchronized final boolean doFlushInternal(boolean flushDocStores, boolean flushDeletes) throws CorruptIndexException, IOException { + private synchronized final boolean doFlushInternal(boolean flushDocStores, boolean flushDeletes, boolean flushToRam) throws CorruptIndexException, IOException { if (hitOOM) { throw new IllegalStateException("this writer hit an OutOfMemoryError; cannot flush"); @@ -4026,7 +4477,7 @@ assert testPoint("startDoFlush"); flushCount++; - + flushDeletes |= docWriter.deletesFull(); // When autoCommit=true we must always flush deletes @@ -4051,7 +4502,22 @@ // Always flush docs if there are any boolean flushDocs = numDocs > 0; - + + // if RAM NRT set the directory to flush to based on the amount of + // ram dir available + if (ramNrt && flushToRam) { + long bytesUsed = docWriter.numBytesUsed; + //long ramDirSize = docWriter.getRamDirSize(); + long ramDirAvail = docWriter.getRamDirAvailable(); + if (bytesUsed >= ramDirAvail) { // no space in the ram dir + docWriter.flushToDir = directory; + } else { + docWriter.flushToDir = switchDirectory; + } + } else { + docWriter.flushToDir = directory; + } + // With autoCommit=true we always must flush the doc // stores when we flush flushDocStores |= autoCommit; @@ -4106,6 +4572,9 @@ try { flushedDocCount = docWriter.flush(flushDocStores); + if (ramNrt) { + docWriter.pushRamDirSize(); + } success = true; } finally { if (!success) { @@ -4129,9 +4598,10 @@ // Create new SegmentInfo, but do not add to our // segmentInfos until deletes are flushed // successfully. + assert docWriter.flushToDir != null; newSegment = new SegmentInfo(segment, flushedDocCount, - directory, false, true, + docWriter.flushToDir, false, true, docStoreOffset, docStoreSegment, docStoreIsCompoundFile, docWriter.hasProx()); @@ -4145,18 +4615,19 @@ if (flushDeletes) { flushDeletesCount++; applyDeletes(); + docWriter.pushRamDirSize(); } - doAfterFlush(); + doAfterFlush(); // pushramdirsize here instead? if (flushDocs) checkpoint(); - if (flushDocs && mergePolicy.useCompoundFile(segmentInfos, newSegment)) { + if (flushDocs && getCurrentMergePolicy().useCompoundFile(segmentInfos, newSegment)) { // Now build compound file boolean success = false; try { - docWriter.createCompoundFile(segment); + docWriter.createCompoundFile(segment, getDirectory()); success = true; } finally { if (!success) { @@ -4184,11 +4655,12 @@ /** Expert: Return the total size of all index files currently cached in memory. * Useful for size management with flushRamDocs() */ + // TODO: do we want this to include the ram dir size? public final long ramSizeInBytes() { ensureOpen(); return docWriter.getRAMUsed(); } - + /** Expert: Return the number of documents currently * buffered in RAM. */ public final synchronized int numRamDocs() { @@ -4200,20 +4672,20 @@ int first = segmentInfos.indexOf(merge.segments.info(0)); if (first == -1) - throw new MergePolicy.MergeException("could not find segment " + merge.segments.info(0).name + " in current index " + segString(), directory); + throw new MergePolicy.MergeException("could not find segment " + merge.segments.info(0).name + " in current index " + segString(), merge.directory); final int numSegments = segmentInfos.size(); final int numSegmentsToMerge = merge.segments.size(); for(int i=0;i= numSegments || !segmentInfos.info(first+i).equals(info)) { if (segmentInfos.indexOf(info) == -1) throw new MergePolicy.MergeException("MergePolicy selected a segment (" + info.name + ") that is not in the current index " + segString(), directory); else - throw new MergePolicy.MergeException("MergePolicy selected non-contiguous segments to merge (" + merge.segString(directory) + " vs " + segString() + "), which IndexWriter (currently) cannot handle", - directory); + throw new MergePolicy.MergeException("MergePolicy selected non-contiguous segments to merge (" + merge.segString() + " vs " + segString() + "), which IndexWriter (currently) cannot handle", + directory); } } @@ -4236,7 +4708,7 @@ final SegmentInfos sourceSegments = merge.segments; if (infoStream != null) - message("commitMergeDeletes " + merge.segString(directory)); + message("commitMergeDeletes " + merge.segString()); // Carefully merge deletes that occurred after we // started merging: @@ -4305,7 +4777,7 @@ } if (infoStream != null) - message("commitMerge: " + merge.segString(directory) + " index=" + segString()); + message("commitMerge: " + merge.segString() + " index=" + segString()); assert merge.registerDone; @@ -4317,7 +4789,7 @@ // abort this merge if (merge.isAborted()) { if (infoStream != null) - message("commitMerge: skipping merge " + merge.segString(directory) + ": it was aborted"); + message("commitMerge: skipping merge " + merge.segString() + ": it was aborted"); deleter.refresh(merge.info.name); return false; @@ -4371,7 +4843,7 @@ final private void handleMergeException(Throwable t, MergePolicy.OneMerge merge) throws IOException { if (infoStream != null) { - message("handleMergeException: merge=" + merge.segString(directory) + " exc=" + t); + message("handleMergeException: merge=" + merge.segString() + " exc=" + t); } // Set the exception on the merge, so if @@ -4409,15 +4881,16 @@ throws CorruptIndexException, IOException { boolean success = false; - + if (ramNrt && merge.directory == switchDirectory) { + assert !merge.useCompoundFile; + } try { try { try { mergeInit(merge); if (infoStream != null) - message("now merge\n merge=" + merge.segString(directory) + "\n merge=" + merge + "\n index=" + segString()); - + message("now merge\n merge=" + merge.segString() + "\n merge=" + merge + "\n index=" + segString()); mergeMiddle(merge); success = true; } catch (Throwable t) { @@ -4432,8 +4905,11 @@ message("hit exception during merge"); if (merge.info != null && !segmentInfos.contains(merge.info)) deleter.refresh(merge.info.name); + } else { + // if the merge succeeded then push the ram dir size + docWriter.pushRamDirSize(); } - + // This merge (and, generally, any change to the // segments) may now enable new merges, so we call // merge policy & update pending merges. @@ -4459,9 +4935,9 @@ if (stopMerges) { merge.abort(); - throw new MergePolicy.MergeAbortedException("merge is aborted: " + merge.segString(directory)); + throw new MergePolicy.MergeAbortedException("merge is aborted: " + merge.segString()); } - + final int count = merge.segments.size(); boolean isExternal = false; for(int i=0;iThis class implements a {@link MergePolicy} that tries @@ -53,21 +55,21 @@ * or larger will never be merged. @see setMaxMergeDocs */ public static final int DEFAULT_MAX_MERGE_DOCS = Integer.MAX_VALUE; - private int mergeFactor = DEFAULT_MERGE_FACTOR; + protected int mergeFactor = DEFAULT_MERGE_FACTOR; long minMergeSize; long maxMergeSize; int maxMergeDocs = DEFAULT_MAX_MERGE_DOCS; - private boolean useCompoundFile = true; - private boolean useCompoundDocStore = true; - private IndexWriter writer; + protected boolean useCompoundFile = true; + protected boolean useCompoundDocStore = true; + protected IndexWriter writer; protected boolean verbose() { return writer != null && writer.verbose(); } - private void message(String message) { + protected void message(String message) { if (verbose()) writer.message("LMP: " + message); } @@ -301,7 +303,7 @@ return spec; } - + /** Checks if any merges are now necessary and returns a * {@link MergePolicy.MergeSpecification} if so. A merge * is necessary when there are more than {@link @@ -309,8 +311,7 @@ * multiple levels have too many segments, this method * will return multiple merges, allowing the {@link * MergeScheduler} to use concurrency. */ - public MergeSpecification findMerges(SegmentInfos infos, IndexWriter writer) throws IOException { - + public MergeSpecification findMerges(SegmentInfos infos, IndexWriter writer) throws IOException { final int numSegments = infos.size(); this.writer = writer; if (verbose()) @@ -321,8 +322,6 @@ float[] levels = new float[numSegments]; final float norm = (float) Math.log(mergeFactor); - final Directory directory = writer.getDirectory(); - for(int i=0;i 0) b.append(' '); - b.append(segments.info(i).segString(dir)); + b.append(segments.info(i).segString(directory)); } if (info != null) b.append(" into ").append(info.name); @@ -154,15 +155,29 @@ public void add(OneMerge merge) { merges.add(merge); } - - public String segString(Directory dir) { + + /** + * @deprecated please call {@link #segString()}) instead as merge has the directory + * @param directory + * @return + */ + public String segString(Directory directory) { StringBuffer b = new StringBuffer(); b.append("MergeSpec:\n"); final int count = merges.size(); for(int i=0;iExpert: {@link IndexWriter} uses an instance * implementing this interface to execute the merges * selected by a {@link MergePolicy}. The default @@ -35,7 +37,7 @@ */ public abstract class MergeScheduler { - + /** Run the merges provided by {@link IndexWriter#getNextMerge()}. */ abstract void merge(IndexWriter writer) throws CorruptIndexException, IOException; Index: src/java/org/apache/lucene/index/MultiSegmentReader.java =================================================================== --- src/java/org/apache/lucene/index/MultiSegmentReader.java (revision 773472) +++ src/java/org/apache/lucene/index/MultiSegmentReader.java (working copy) @@ -83,14 +83,16 @@ // no need to process segments in reverse order final int numSegments = infos.size(); SegmentReader[] readers = new SegmentReader[numSegments]; - final Directory dir = writer.getDirectory(); + final Directory flushDir = writer.getInternalDirectory(); int upto = 0; for (int i=0;i= writer.docWriter.getRamDirSize(); + } + + private static long getSize(MergePolicy.OneMerge merge) throws IOException { + long total = 0; + for (int x = 0; x < merge.segments.size(); x++) { + SegmentInfo si = (SegmentInfo) merge.segments.get(x); + total += si.sizeInBytes(); + } + return total; + } + + private static void setDir(MergeSpecification spec, Directory dir) { + if (spec != null) { + for (int x = 0; x < spec.merges.size(); x++) { + MergePolicy.OneMerge merge = (MergePolicy.OneMerge) spec.merges.get(x); + merge.directory = dir; + } + } + } + + private static MergeSpecification mergeSpecs(MergeSpecification ms1, MergeSpecification ms2) { + if (ms1 == null && ms2 == null) return null; + MergeSpecification ms = new MergeSpecification(); + if (ms1 != null) ms.merges.addAll(ms1.merges); + if (ms2 != null) ms.merges.addAll(ms2.merges); + return ms; + } +} Index: src/java/org/apache/lucene/index/RAMMergePolicy.java =================================================================== --- src/java/org/apache/lucene/index/RAMMergePolicy.java (revision 0) +++ src/java/org/apache/lucene/index/RAMMergePolicy.java (revision 0) @@ -0,0 +1,25 @@ +package org.apache.lucene.index; + +/** + * Merge Policy for merging IndexWriter ram segments + */ +public class RAMMergePolicy extends LogDocMergePolicy { + + public RAMMergePolicy() { + super(); + useCompoundFile = false; + useCompoundDocStore = false; + } + + protected long size(SegmentInfo info) { + return info.docCount; + } + + public boolean getUseCompoundFile() { + return false; + } + + public boolean useCompoundDocStore(SegmentInfos infos) { + return false; + } +} Index: src/java/org/apache/lucene/index/SegmentInfos.java =================================================================== --- src/java/org/apache/lucene/index/SegmentInfos.java (revision 773472) +++ src/java/org/apache/lucene/index/SegmentInfos.java (working copy) @@ -812,12 +812,27 @@ final int size = size(); for(int i=0;i= 10000.0) { - merge.checkAborted(dir); + merge.checkAborted(merge.directory); workCount = 0; } } Index: src/java/org/apache/lucene/index/SerialMergeScheduler.java =================================================================== --- src/java/org/apache/lucene/index/SerialMergeScheduler.java (revision 773472) +++ src/java/org/apache/lucene/index/SerialMergeScheduler.java (working copy) @@ -19,16 +19,17 @@ import java.io.IOException; +import org.apache.lucene.store.Directory; + /** A {@link MergeScheduler} that simply does each merge * sequentially, using the current thread. */ -public class SerialMergeScheduler extends MergeScheduler { - +public class SerialMergeScheduler extends MergeScheduler { /** Just do the merges in sequence. We do this * "synchronized" so that even if the application is using * multiple threads, only one merge may run at a time. */ synchronized public void merge(IndexWriter writer) throws CorruptIndexException, IOException { - + while(true) { MergePolicy.OneMerge merge = writer.getNextMerge(); if (merge == null) @@ -36,6 +37,6 @@ writer.merge(merge); } } - + public void close() {} } Index: src/java/org/apache/lucene/index/StoredFieldsWriter.java =================================================================== --- src/java/org/apache/lucene/index/StoredFieldsWriter.java (revision 773472) +++ src/java/org/apache/lucene/index/StoredFieldsWriter.java (working copy) @@ -20,6 +20,7 @@ import java.util.Map; import java.io.IOException; import org.apache.lucene.store.RAMOutputStream; +import org.apache.lucene.store.Directory; import org.apache.lucene.util.ArrayUtil; /** This is a DocFieldConsumer that writes stored fields. */ @@ -46,7 +47,7 @@ // It's possible that all documents seen in this segment // hit non-aborting exceptions, in which case we will // not have yet init'd the FieldsWriter: - initFieldsWriter(); + initFieldsWriter(state.directory); // Fill fdx file to include any final docs that we // skipped because they hit non-aborting exceptions @@ -57,12 +58,12 @@ fieldsWriter.flush(); } - private void initFieldsWriter() throws IOException { + private void initFieldsWriter(Directory directory) throws IOException { if (fieldsWriter == null) { final String docStoreSegment = docWriter.getDocStoreSegment(); if (docStoreSegment != null) { assert docStoreSegment != null; - fieldsWriter = new FieldsWriter(docWriter.directory, + fieldsWriter = new FieldsWriter(directory, docStoreSegment, fieldInfos); docWriter.addOpenFile(docStoreSegment + "." + IndexFileNames.FIELDS_EXTENSION); @@ -75,7 +76,7 @@ synchronized public void closeDocStore(SegmentWriteState state) throws IOException { final int inc = state.numDocsInStore - lastDocID; if (inc > 0) { - initFieldsWriter(); + initFieldsWriter(state.directory); fill(state.numDocsInStore - docWriter.getDocStoreOffset()); } @@ -138,7 +139,7 @@ synchronized void finishDocument(PerDoc perDoc) throws IOException { assert docWriter.writer.testPoint("StoredFieldsWriter.finishDocument start"); - initFieldsWriter(); + initFieldsWriter(docWriter.writer.getDirectory()); // TODO: we need a way to pass the directory in here from segmentWriteState fill(perDoc.docID); Index: src/java/org/apache/lucene/index/TermVectorsTermsWriter.java =================================================================== --- src/java/org/apache/lucene/index/TermVectorsTermsWriter.java (revision 773472) +++ src/java/org/apache/lucene/index/TermVectorsTermsWriter.java (working copy) @@ -19,6 +19,7 @@ import org.apache.lucene.store.IndexOutput; import org.apache.lucene.store.RAMOutputStream; +import org.apache.lucene.store.Directory; import org.apache.lucene.util.ArrayUtil; import java.io.IOException; @@ -36,9 +37,12 @@ IndexOutput tvd; IndexOutput tvf; int lastDocID; + Directory dir; public TermVectorsTermsWriter(DocumentsWriter docWriter) { this.docWriter = docWriter; + // term vectors always go to the primary directory + this.dir = docWriter.writer.getDirectory(); } public TermsHashConsumerPerThread addThread(TermsHashPerThread termsHashPerThread) { @@ -152,9 +156,9 @@ // vector output files, we must abort this segment // because those files will be in an unknown // state: - tvx = docWriter.directory.createOutput(docStoreSegment + "." + IndexFileNames.VECTORS_INDEX_EXTENSION); - tvd = docWriter.directory.createOutput(docStoreSegment + "." + IndexFileNames.VECTORS_DOCUMENTS_EXTENSION); - tvf = docWriter.directory.createOutput(docStoreSegment + "." + IndexFileNames.VECTORS_FIELDS_EXTENSION); + tvx = dir.createOutput(docStoreSegment + "." + IndexFileNames.VECTORS_INDEX_EXTENSION); + tvd = dir.createOutput(docStoreSegment + "." + IndexFileNames.VECTORS_DOCUMENTS_EXTENSION); + tvf = dir.createOutput(docStoreSegment + "." + IndexFileNames.VECTORS_FIELDS_EXTENSION); tvx.writeInt(TermVectorsReader.FORMAT_CURRENT); tvd.writeInt(TermVectorsReader.FORMAT_CURRENT); Index: src/java/org/apache/lucene/store/Directory.java =================================================================== --- src/java/org/apache/lucene/store/Directory.java (revision 772407) +++ src/java/org/apache/lucene/store/Directory.java (working copy) @@ -67,7 +67,17 @@ { return list(); } - + + /** + * Returns true if the given directory is equal to or + * is contained within this directory. + * @param other Directory to compare against + * @return if this directory equals or contains the given directory + */ + public boolean contains(Directory other) { + return equals(other); + } + /** Returns true iff a file with the given name exists. */ public abstract boolean fileExists(String name) throws IOException; Index: src/java/org/apache/lucene/store/FileSwitchDirectory.java =================================================================== --- src/java/org/apache/lucene/store/FileSwitchDirectory.java (revision 772407) +++ src/java/org/apache/lucene/store/FileSwitchDirectory.java (working copy) @@ -44,6 +44,13 @@ this.lockFactory = primaryDir.getLockFactory(); } + public boolean contains(Directory other) { + if (other instanceof FileSwitchDirectory) { + return super.contains(other); + } + return primaryDir.equals(other) || secondaryDir.equals(other); + } + public Directory getPrimaryDir() { return primaryDir; } @@ -84,7 +91,9 @@ return name.substring(i+1, name.length()); } - private Directory getDirectory(String name) { + private Directory getDirectory(String name) throws IOException { + if (primaryDir.fileExists(name)) return primaryDir; + if (secondaryDir.fileExists(name)) return secondaryDir; String ext = getExtension(name); if (primaryExtensions.contains(ext)) { return primaryDir; Index: src/java/org/apache/lucene/store/RAMDirectory.java =================================================================== --- src/java/org/apache/lucene/store/RAMDirectory.java (revision 772407) +++ src/java/org/apache/lucene/store/RAMDirectory.java (working copy) @@ -37,7 +37,7 @@ private static final long serialVersionUID = 1l; HashMap fileMap = new HashMap(); - long sizeInBytes = 0; + volatile long sizeInBytes = 0; // ***** // Lock acquisition sequence: RAMDirectory, then RAMFile Index: src/test/org/apache/lucene/index/TestCrash.java =================================================================== --- src/test/org/apache/lucene/index/TestCrash.java (revision 772355) +++ src/test/org/apache/lucene/index/TestCrash.java (working copy) @@ -49,7 +49,7 @@ return writer; } - private void crash(final IndexWriter writer) throws IOException { + public static void crash(final IndexWriter writer) throws IOException { final MockRAMDirectory dir = (MockRAMDirectory) writer.getDirectory(); ConcurrentMergeScheduler cms = (ConcurrentMergeScheduler) writer.getMergeScheduler(); dir.crash(); Index: src/test/org/apache/lucene/index/TestIndexWriterRamDir.java =================================================================== --- src/test/org/apache/lucene/index/TestIndexWriterRamDir.java (revision 0) +++ src/test/org/apache/lucene/index/TestIndexWriterRamDir.java (revision 0) @@ -0,0 +1,608 @@ +package org.apache.lucene.index; + +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Set; + +import org.apache.lucene.analysis.WhitespaceAnalyzer; +import org.apache.lucene.document.Document; +import org.apache.lucene.store.Directory; +import org.apache.lucene.store.FSDirectory; +import org.apache.lucene.store.FileSwitchDirectory; +import org.apache.lucene.store.MockRAMDirectory; +import org.apache.lucene.store.RAMDirectory; +import org.apache.lucene.util.LuceneTestCase; + +/** + * Tests the IndexWriter ram directory where if a ram directory is + * set on the writer, flushes occur to it first, then are later + * merged to disk. + */ +public class TestIndexWriterRamDir extends LuceneTestCase { + + private static MockRAMDirectory createRamDir() throws IOException { + Set prevExts = new HashSet(); + prevExts.add("cfs"); // prevent compound files + MockRAMDirectory ramDir = new MockRAMDirectory(prevExts); + return ramDir; + } + + /** + * Test if IW.optimize merges all segments into 1 and + * moves them to disk. + * @throws Exception + */ + public void testOptimize() throws Exception { + Directory dir = new MockRAMDirectory(); + MockRAMDirectory ramDir = createRamDir(); + // have ramdir assert compound files aren't created + IndexWriter writer = new IndexWriter(dir, new WhitespaceAnalyzer(), + IndexWriter.MaxFieldLength.LIMITED, true, ramDir); + assertTrue(writer.ramNrt); + ConcurrentMergeScheduler cms = new ConcurrentMergeScheduler(); + writer.setMergeScheduler(cms); + // create some segments + int rc = 0; + for (int x=0; x < 9000; x++) { + Document d = TestIndexWriterReader.createDocument(x, "primary", 5); + writer.addDocument(d); + if (x % 500 == 0) { + writer.flush(); + rc++; + } + } + writer.commit(); // commit them to RAM + for (int x=0; x < 4000; x++) { + Document d = TestIndexWriterReader.createDocument(x, "primary", 5); + writer.addDocument(d); + if (x % 500 == 0) writer.flush(); + } + writer.optimize(); + cms.sync(); + IndexReader reader = writer.getReader(); + IndexReader[] readers = getAllReaders(reader); + assertEquals(1, readers.length); // insure we have only 1 segment on disk + SegmentInfos infos = writer.getSegmentInfos(); + assertEquals(1, infos.size()); + SegmentInfo info = infos.info(0); + assertEquals(dir, info.dir); // verify the directory is the primary + reader.close(); + writer.close(); + dir.close(); + } + + /** + * Test to insure expungeDeletes actually removes all + * segments with deleted docs when RAM NRT is on. + * @throws Exception + */ + public void testExpungeDeletes() throws Exception { + Directory dir = new MockRAMDirectory(); + IndexWriter writer = new IndexWriter(dir, new WhitespaceAnalyzer(), + IndexWriter.MaxFieldLength.LIMITED, true); + assertTrue(writer.ramNrt); + ConcurrentMergeScheduler cms = new ConcurrentMergeScheduler(); + writer.setMergeScheduler(cms); + // create some segments + for (int x=0; x < 9000; x++) { + Document d = TestIndexWriterReader.createDocument(x, "primary", 5); + writer.addDocument(d); + if (x % 500 == 0) writer.flush(); + } + writer.commit(); // commit them to RAM + for (int x=0; x < 9000; x++) { + Document d = TestIndexWriterReader.createDocument(x, "primary", 5); + writer.addDocument(d); + if (x % 500 == 0) writer.flush(); + } + writer.expungeDeletes(); + cms.sync(); + IndexReader reader = writer.getReader(); + IndexReader[] readers = getAllReaders(reader); + for (int x=0; x < readers.length; x++) { + assertEquals(0, readers[x].numDeletedDocs()); + } + reader.close(); + writer.close(); + dir.close(); + } + + static private final IndexReader[] indexReaderZeroArray = new IndexReader[0]; + + private static void gatherSubReaders(List allSubReaders, IndexReader r) { + IndexReader[] subReaders = r.getSequentialSubReaders(); + if (subReaders == null) { + // Add the reader itself, and do not recurse + allSubReaders.add(r); + } else { + for(int i=0;i 0); + IndexReader reader = writer.getReader(); + assertEquals(5000, reader.maxDoc()); + //System.out.println("infos.size:"+infos.size()+" raminfos.size:"+writer.getRamDirSegmentInfos().size()); + reader.close(); + writer.close(); + dir.close(); + } + + /** + * Test IndexWriter doing in ram merges + * @throws IOException + */ + public void testMergeInRam() throws IOException { + Directory dir = new MockRAMDirectory(); + IndexWriter writer = new IndexWriter(dir, new WhitespaceAnalyzer(), + IndexWriter.MaxFieldLength.LIMITED, true); + ConcurrentMergeScheduler cms = new ConcurrentMergeScheduler(); + writer.setMergeScheduler(cms); + + LogDocMergePolicy mp = new LogDocMergePolicy(); + writer.setMergePolicy(mp); + + RAMMergePolicy ramMp = new RAMMergePolicy(); + ramMp.setMergeFactor(2); + writer.setRAMMergePolicy(ramMp); + + // create 10 segments with 500 documents each + for (int x=0; x < 5000; x++) { + Document d = TestIndexWriterReader.createDocument(x, "ram", 5); + writer.addDocument(d); + if (x % 500 == 0) { + writer.flush(); + } + } + SegmentInfos infos = writer.getSegmentInfos(); + assertTrue(writer.getRamDirSegmentInfos().size() > 0); + IndexReader reader = writer.getReader(); + assertEquals(5000, reader.maxDoc()); + //System.out.println("infos.size:"+infos.size()+" raminfos.size:"+writer.getRamDirSegmentInfos().size()); + reader.close(); + writer.close(); + dir.close(); + } + + public static class NoMergeScheduler extends MergeScheduler { + public void merge(IndexWriter writer) throws CorruptIndexException, IOException {} + public void close() {} + } + + /** + * Make sure IW.commit flushes all ram segments + * to the directory + * + * @throws IOException + */ + public void testCommit() throws IOException { + Directory dir = new MockRAMDirectory(); + IndexWriter writer = new IndexWriter(dir, new WhitespaceAnalyzer(), + IndexWriter.MaxFieldLength.LIMITED, true); + ConcurrentMergeScheduler cms = new ConcurrentMergeScheduler(); + writer.setMergeScheduler(cms); + //writer.setMergeScheduler(new NoMergeScheduler()); + + LogDocMergePolicy mp = new LogDocMergePolicy(); + writer.setMergePolicy(mp); + + // create 10 segments with 500 documents each + for (int x=0; x < 5000; x++) { + Document d = TestIndexWriterReader.createDocument(x, "ram", 5); + writer.addDocument(d); + if (x % 500 == 0) { + writer.flush(); + } + } + SegmentInfos infos = writer.getSegmentInfos(); + assertTrue(infos.size() == 10); + assertTrue(hasDirSegmentInfos(writer.getInternalDirectory(), infos)); + writer.resolveRamSegments(); // should happen in the foreground + SegmentInfos ramInfos = writer.getRamDirSegmentInfos(); + assertEquals(0, ramInfos.size()); + + IndexReader reader = writer.getReader(); + assertEquals(5000, reader.maxDoc()); + + reader.close(); + writer.close(); + dir.close(); + } + + /** + * Test IndexWriter merging to disk + * @throws IOException + */ + public void testMergeToPrimaryDir() throws IOException { + Directory dir = new MockRAMDirectory(); + IndexWriter writer = new IndexWriter(dir, new WhitespaceAnalyzer(), + IndexWriter.MaxFieldLength.LIMITED, true); + ConcurrentMergeScheduler cms = new ConcurrentMergeScheduler(); + writer.setMergeScheduler(cms); + + LogDocMergePolicy mp = new LogDocMergePolicy(); + writer.setMergePolicy(mp); + + // create 10 segments with 500 documents each + for (int x=0; x < 5000; x++) { + Document d = TestIndexWriterReader.createDocument(x, "ram", 5); + writer.addDocument(d); + if (x % 500 == 0) { + writer.flush(); + } + } + SegmentInfos infos = writer.getSegmentInfos(); + //System.out.println("infos.size:"+infos.size()+" raminfos.size:"+writer.getRamDirSegmentInfos().size()); + assertTrue(infos.size() == 10); + assertTrue(hasDirSegmentInfos(writer.getInternalDirectory(), infos)); + writer.mergeRamSegmentsToDir(); + cms.sync(); // wait for the merges to complete + //System.out.println("ram infos.size:"+writer.getRamDirSegmentInfos().size()+" "+IndexWriter.getSegmentNames(writer.getRamDirSegmentInfos())); + // there shouldn't be any ram segments + SegmentInfos ramInfos = writer.getRamDirSegmentInfos(); + // make sure the number of segments decreased + assertTrue(ramInfos.size() < 3); + + IndexReader reader = writer.getReader(); + assertEquals(5000, reader.maxDoc()); + + reader.close(); + writer.close(); + dir.close(); + } + + private static boolean hasDirSegmentInfos(Directory dir, SegmentInfos infos) { + int dirSegs = 0; + for (int x=0; x < infos.size(); x++) { + SegmentInfo info = infos.info(x); + if (info.dir == dir) { + dirSegs++; + } + } + return dirSegs > 0; + } + + /** + * Test the ram dir is working and that the + * reader returned from IndexWriter encompasses + * the in ram segments. + * @throws IOException + */ + public void testRamDir() throws IOException { + Directory dir = new MockRAMDirectory(); + IndexWriter writer = new IndexWriter(dir, new WhitespaceAnalyzer(), + IndexWriter.MaxFieldLength.LIMITED, true); + for (int x=0; x < 100; x++) { + Document d = TestIndexWriterReader.createDocument(x, "ram", 5); + writer.addDocument(d); + } + writer.flush(false, false, true); + SegmentInfos infos = writer.getSegmentInfos(); + assertTrue(writer.getDirectory() != writer.getInternalDirectory()); + boolean hasRamDirSegs = hasDirSegmentInfos(writer.getInternalDirectory(), infos); + //System.out.println("ramDirSegs:"+ramDirSegs); + assertTrue(hasRamDirSegs); + IndexReader ramReader = writer.getReader(); + assertEquals(100, ramReader.maxDoc()); + writer.close(); + ramReader.close(); + dir.close(); + } +} Index: src/test/org/apache/lucene/store/MockRAMDirectory.java =================================================================== --- src/test/org/apache/lucene/store/MockRAMDirectory.java (revision 772355) +++ src/test/org/apache/lucene/store/MockRAMDirectory.java (working copy) @@ -47,6 +47,7 @@ private Set unSyncedFiles; private Set createdFiles; volatile boolean crashed; + private Set extPrevent; // NOTE: we cannot initialize the Map here due to the // order in which our constructor actually does this @@ -79,6 +80,11 @@ super(dir); init(); } + + public MockRAMDirectory(Set extPrevent) throws IOException { + this.extPrevent = extPrevent; + init(); + } /** If set to true, we throw an IOException if the same * file is opened by createOutput, ever. */ @@ -203,11 +209,19 @@ } super.deleteFile(name); } - + + private boolean extensionOK(String name) { + if (extPrevent == null) return true; + String ext = FileSwitchDirectory.getExtension(name); + if (ext == null || ext.equals("")) return false; + return !extPrevent.contains(ext); + } + public IndexOutput createOutput(String name) throws IOException { if (crashed) throw new IOException("cannot createOutput after crash"); init(); + assert extensionOK(name); synchronized(openFiles) { if (preventDoubleWrite && createdFiles.contains(name) && !name.equals("segments.gen")) throw new IOException("file \"" + name + "\" was already written to"); Index: src/test/org/apache/lucene/store/TestMockRAMDirectory.java =================================================================== --- src/test/org/apache/lucene/store/TestMockRAMDirectory.java (revision 0) +++ src/test/org/apache/lucene/store/TestMockRAMDirectory.java (revision 0) @@ -0,0 +1,23 @@ +package org.apache.lucene.store; + +import java.io.IOException; +import java.util.HashSet; +import java.util.Set; + +import org.apache.lucene.util.LuceneTestCase; + +public class TestMockRAMDirectory extends LuceneTestCase { + public void testMockRAMDirectory() throws IOException { + Set set = new HashSet(); + set.add("cfs"); + MockRAMDirectory ramDir = new MockRAMDirectory(set); + boolean ex = false; + try { + IndexOutput output = ramDir.createOutput("test1.cfs"); + } catch (Throwable th) { + ex = true; + } + ramDir.close(); + assertTrue(ex); + } +}