Index: CHANGES.txt =================================================================== --- CHANGES.txt (revision 770782) +++ CHANGES.txt (working copy) @@ -283,6 +283,12 @@ 22. LUCENE-1605: Added BitVector.subset(). (Jeremy Volkman via Mike McCandless) +23. 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 FSDirectory. + (Jason Rutherglen via Mike McCandless) + Optimizations 1. LUCENE-1427: Fixed QueryWrapperFilter to not waste time computing Index: src/java/org/apache/lucene/index/DirectoryIndexReader.java =================================================================== --- src/java/org/apache/lucene/index/DirectoryIndexReader.java (revision 770782) +++ src/java/org/apache/lucene/index/DirectoryIndexReader.java (working copy) @@ -400,7 +400,7 @@ segmentInfos.setUserData(commitUserData); // Default deleter (for backwards compatibility) is // KeepOnlyLastCommitDeleter: - IndexFileDeleter deleter = new IndexFileDeleter(directory, + IndexFileDeleter deleter = new IndexFileDeleter(directory, null, deletionPolicy == null ? new KeepOnlyLastCommitDeletionPolicy() : deletionPolicy, segmentInfos, null, null); Index: src/java/org/apache/lucene/index/SegmentInfo.java =================================================================== --- src/java/org/apache/lucene/index/SegmentInfo.java (revision 770782) +++ src/java/org/apache/lucene/index/SegmentInfo.java (working copy) @@ -549,7 +549,7 @@ * modify it. */ - public List files() throws IOException { + public synchronized List files() throws IOException { if (files != null) { // Already cached: @@ -644,7 +644,7 @@ /* Called whenever any change is made that affects which * files this segment has. */ - private void clearFiles() { + private synchronized void clearFiles() { files = null; sizeInBytes = -1; } Index: src/java/org/apache/lucene/index/DocFieldProcessorPerThread.java =================================================================== --- src/java/org/apache/lucene/index/DocFieldProcessorPerThread.java (revision 770782) +++ 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,7 @@ } public DocumentsWriter.DocWriter processDocument() throws IOException { - + Directory dir = this.docState.docWriter.writer.getFlushDirectory(); consumer.startDocument(); final Document doc = docState.doc; Index: src/java/org/apache/lucene/index/MergePolicy.java =================================================================== --- src/java/org/apache/lucene/index/MergePolicy.java (revision 770782) +++ src/java/org/apache/lucene/index/MergePolicy.java (working copy) @@ -85,6 +85,7 @@ final boolean useCompoundFile; boolean aborted; Throwable error; + Directory directory; public OneMerge(SegmentInfos segments, boolean useCompoundFile) { if (0 == segments.size()) Index: src/java/org/apache/lucene/index/LogMergePolicy.java =================================================================== --- src/java/org/apache/lucene/index/LogMergePolicy.java (revision 770782) +++ src/java/org/apache/lucene/index/LogMergePolicy.java (working copy) @@ -19,7 +19,9 @@ import java.io.IOException; import java.util.Set; +import java.util.List; +import org.apache.lucene.index.MergePolicy.OneMerge; import org.apache.lucene.store.Directory; /**

This class implements a {@link MergePolicy} that tries @@ -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;iGet the current setting of whether newly flushed * segments will use the compound file format. Note that @@ -847,7 +874,7 @@ */ public IndexWriter(String path, Analyzer a, boolean create, MaxFieldLength mfl) throws CorruptIndexException, LockObtainFailedException, IOException { - init(FSDirectory.getDirectory(path), a, create, true, null, false, mfl.getLimit(), null, null); + init(FSDirectory.getDirectory(path), null, a, create, true, null, false, mfl.getLimit(), null, null); } /** @@ -876,7 +903,7 @@ */ public IndexWriter(String path, Analyzer a, boolean create) throws CorruptIndexException, LockObtainFailedException, IOException { - init(FSDirectory.getDirectory(path), a, create, true, null, true, DEFAULT_MAX_FIELD_LENGTH, null, null); + init(FSDirectory.getDirectory(path), null, a, create, true, null, true, DEFAULT_MAX_FIELD_LENGTH, null, null); } /** @@ -907,7 +934,7 @@ */ public IndexWriter(File path, Analyzer a, boolean create, MaxFieldLength mfl) throws CorruptIndexException, LockObtainFailedException, IOException { - init(FSDirectory.getDirectory(path), a, create, true, null, false, mfl.getLimit(), null, null); + init(FSDirectory.getDirectory(path), null, a, create, true, null, false, mfl.getLimit(), null, null); } /** @@ -936,7 +963,7 @@ */ public IndexWriter(File path, Analyzer a, boolean create) throws CorruptIndexException, LockObtainFailedException, IOException { - init(FSDirectory.getDirectory(path), a, create, true, null, true, DEFAULT_MAX_FIELD_LENGTH, null, null); + init(FSDirectory.getDirectory(path), null, a, create, true, null, true, DEFAULT_MAX_FIELD_LENGTH, null, null); } /** @@ -967,7 +994,7 @@ */ public IndexWriter(Directory d, Analyzer a, boolean create, MaxFieldLength mfl) throws CorruptIndexException, LockObtainFailedException, IOException { - init(d, a, create, false, null, false, mfl.getLimit(), null, null); + init(d, null, a, create, false, null, false, mfl.getLimit(), null, null); } /** @@ -995,7 +1022,7 @@ */ public IndexWriter(Directory d, Analyzer a, boolean create) throws CorruptIndexException, LockObtainFailedException, IOException { - init(d, a, create, false, null, true, DEFAULT_MAX_FIELD_LENGTH, null, null); + init(d, null, a, create, false, null, true, DEFAULT_MAX_FIELD_LENGTH, null, null); } /** @@ -1126,6 +1153,34 @@ 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 ramDir ram 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. + * @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 + */ + public IndexWriter(Directory d, Directory ramDir, Analyzer a, MaxFieldLength mfl) + throws CorruptIndexException, LockObtainFailedException, IOException { + init(d, ramDir, a, false, null, false, mfl.getLimit(), null, null); + } /** * Constructs an IndexWriter for the index in @@ -1205,7 +1260,7 @@ */ public IndexWriter(Directory d, boolean autoCommit, Analyzer a, boolean create) throws CorruptIndexException, LockObtainFailedException, IOException { - init(d, a, create, false, null, autoCommit, DEFAULT_MAX_FIELD_LENGTH, null, null); + init(d, null, a, create, false, null, autoCommit, DEFAULT_MAX_FIELD_LENGTH, null, null); } /** @@ -1292,7 +1347,7 @@ */ public IndexWriter(Directory d, Analyzer a, boolean create, IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl) throws CorruptIndexException, LockObtainFailedException, IOException { - init(d, a, create, false, deletionPolicy, false, mfl.getLimit(), null, null); + init(d, null, a, create, false, deletionPolicy, false, mfl.getLimit(), null, null); } /** @@ -1329,7 +1384,7 @@ */ IndexWriter(Directory d, Analyzer a, boolean create, IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl, IndexingChain indexingChain, IndexCommit commit) throws CorruptIndexException, LockObtainFailedException, IOException { - init(d, a, create, false, deletionPolicy, false, mfl.getLimit(), indexingChain, commit); + init(d, null, a, create, false, deletionPolicy, false, mfl.getLimit(), indexingChain, commit); } /** @@ -1362,9 +1417,9 @@ */ public IndexWriter(Directory d, boolean autoCommit, Analyzer a, boolean create, IndexDeletionPolicy deletionPolicy) throws CorruptIndexException, LockObtainFailedException, IOException { - init(d, a, create, false, deletionPolicy, autoCommit, DEFAULT_MAX_FIELD_LENGTH, null, null); + init(d, null, a, create, false, deletionPolicy, autoCommit, DEFAULT_MAX_FIELD_LENGTH, null, null); } - + /** * Expert: constructs an IndexWriter on specific commit * point, with a custom {@link IndexDeletionPolicy}, for @@ -1403,25 +1458,78 @@ */ public IndexWriter(Directory d, Analyzer a, IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl, IndexCommit commit) throws CorruptIndexException, LockObtainFailedException, IOException { - init(d, a, false, false, deletionPolicy, false, mfl.getLimit(), null, commit); + init(d, null, a, false, false, deletionPolicy, false, mfl.getLimit(), null, commit); } + + /** + * Expert: constructs an IndexWriter on specific commit + * point, with a custom {@link IndexDeletionPolicy}, for + * the index in d. Text will be analyzed + * with a. + * + *

This is only meaningful if you've used a {@link + * IndexDeletionPolicy} in that past that keeps more than + * just the last commit. + * + *

This operation is similar to {@link #rollback()}, + * except that method can only rollback what's been done + * with the current instance of IndexWriter since its last + * commit, whereas this method can rollback to an + * arbitrary commit point from the past, assuming the + * {@link IndexDeletionPolicy} has preserved past + * commits. + * + *

NOTE: autoCommit (see above) is set to false with this + * constructor. + * + * @param d the index directory + * @param ramDirectory ram directory + * @param a the analyzer to use + * @param deletionPolicy see above + * @param mfl whether or not to limit field lengths, value is in number of terms/tokens. See {@link org.apache.lucene.index.IndexWriter.MaxFieldLength}. + * @param commit which commit to open + * @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(Directory d, Directory ramDirectory, Analyzer a, IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl, IndexCommit commit) + throws CorruptIndexException, LockObtainFailedException, IOException { + init(d, ramDirectory, 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) + throws CorruptIndexException, LockObtainFailedException, IOException { + init(d, null, a, closeDir, deletionPolicy, autoCommit, maxFieldLength, indexingChain, commit); + } + + private void init(Directory d, Directory ramDirectory, Analyzer a, boolean closeDir, IndexDeletionPolicy deletionPolicy, boolean autoCommit, int maxFieldLength, IndexingChain indexingChain, IndexCommit commit) throws CorruptIndexException, LockObtainFailedException, IOException { if (IndexReader.indexExists(d)) { - init(d, a, false, closeDir, deletionPolicy, autoCommit, maxFieldLength, indexingChain, commit); + init(d, ramDirectory, a, false, closeDir, deletionPolicy, autoCommit, maxFieldLength, indexingChain, commit); } else { - init(d, a, true, closeDir, deletionPolicy, autoCommit, maxFieldLength, indexingChain, commit); + init(d, ramDirectory, a, true, closeDir, deletionPolicy, autoCommit, maxFieldLength, indexingChain, commit); } } - private void init(Directory d, Analyzer a, final boolean create, boolean closeDir, + private void init(Directory d, Directory ramDirectory, Analyzer a, final boolean create, boolean closeDir, IndexDeletionPolicy deletionPolicy, boolean autoCommit, int maxFieldLength, IndexingChain indexingChain, IndexCommit commit) throws CorruptIndexException, LockObtainFailedException, IOException { this.closeDir = closeDir; directory = d; + if (ramDirectory != null) { + this.switchDirectory = new FileSwitchDirectory(SWITCH_FILE_EXTS, directory, ramDirectory, false); + getRamLogMergePolicy().setUseCompoundDocStore(false); + getRamLogMergePolicy().setUseCompoundFile(false); + } analyzer = a; setMessageID(defaultInfoStream); this.maxFieldLength = maxFieldLength; @@ -1493,13 +1601,13 @@ this.autoCommit = autoCommit; setRollbackSegmentInfos(segmentInfos); - docWriter = new DocumentsWriter(directory, this, indexingChain); + docWriter = new DocumentsWriter(this, indexingChain); docWriter.setInfoStream(infoStream); docWriter.setMaxFieldLength(maxFieldLength); // Default deleter (for backwards compatibility) is // KeepOnlyLastCommitDeleter: - deleter = new IndexFileDeleter(directory, + deleter = new IndexFileDeleter(directory, switchDirectory, deletionPolicy == null ? new KeepOnlyLastCommitDeletionPolicy() : deletionPolicy, segmentInfos, infoStream, docWriter); @@ -1523,16 +1631,42 @@ throw e; } } - + + public Directory getFlushDirectory() { + if (switchDirectory != null) { + return switchDirectory; + } else return directory; + } + + public Directory getRAMDirectory() { + return switchDirectory.getOtherDir(); + } + private synchronized void setRollbackSegmentInfos(SegmentInfos infos) { rollbackSegmentInfos = (SegmentInfos) infos.clone(); - assert !rollbackSegmentInfos.hasExternalSegments(directory); + assert !rollbackSegmentInfos.hasExternalSegments(directory, getFlushDirectory()); rollbackSegments = new HashMap(); final int size = rollbackSegmentInfos.size(); for(int i=0;i getRamBufferSizeLong())) { + // call updatePendingMerges, optimize the ramInfos to directory + updatePendingMerges(1, true, mergePolicy, ramInfos, directory); + updatePendingMerges(maxNumSegmentsOptimize, optimize, mergePolicy, segmentInfos, directory); + } else { + updatePendingMerges(maxNumSegmentsOptimize, optimize, mergePolicy, segmentInfos, directory); + updatePendingMerges(maxNumSegmentsOptimize, optimize, ramMergePolicy, ramInfos, switchDirectory); + } + } else { + updatePendingMerges(maxNumSegmentsOptimize, optimize, mergePolicy, segmentInfos, directory); + } + if (forceToDir) { // this is kind of a hack + mergeScheduler.merge(this); + } + } + + static long size(SegmentInfos infos) throws IOException { + long size = 0; + for (int x=0; x < infos.size(); x++) { + size += infos.info(x).sizeInBytes(); + } + return size; + } + + private synchronized void updatePendingMerges(int maxNumSegmentsOptimize, boolean optimize, MergePolicy mergePolicy, SegmentInfos segmentInfos, Directory directory) throws CorruptIndexException, IOException { assert !optimize || maxNumSegmentsOptimize > 0; @@ -2928,8 +3102,11 @@ if (spec != null) { final int numMerges = spec.merges.size(); - for(int i=0;i= numSegments || !segmentInfos.info(first+i).equals(info)) { + + boolean n1 = first + i >= numSegments; + boolean n2 = !segmentInfos.info(first+i).equals(info); + if (n1 || n2) { + //if (first + 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); + throw new MergePolicy.MergeException("MergePolicy selected a segment (" + info.name + ") that is not in the current index " + segString(), merge.directory); else - throw new MergePolicy.MergeException("MergePolicy selected non-contiguous segments to merge (" + merge.segString(directory) + " vs " + segString() + "), which IndexWriter (currently) cannot handle", + throw new MergePolicy.MergeException("n1: "+n1+" n2:"+n2+" MergePolicy selected non-contiguous segments to merge (" + merge.segString(merge.directory) + " vs " + segString() + "), which IndexWriter (currently) cannot handle", directory); } } @@ -4236,7 +4433,7 @@ final SegmentInfos sourceSegments = merge.segments; if (infoStream != null) - message("commitMergeDeletes " + merge.segString(directory)); + message("commitMergeDeletes " + merge.segString(merge.directory)); // Carefully merge deletes that occurred after we // started merging: @@ -4305,7 +4502,7 @@ } if (infoStream != null) - message("commitMerge: " + merge.segString(directory) + " index=" + segString()); + message("commitMerge: " + merge.segString(merge.directory) + " index=" + segString()); assert merge.registerDone; @@ -4317,7 +4514,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(merge.directory) + ": it was aborted"); deleter.refresh(merge.info.name); return false; @@ -4437,8 +4634,9 @@ // This merge (and, generally, any change to the // segments) may now enable new merges, so we call // merge policy & update pending merges. - if (success && !merge.isAborted() && !closed && !closing) - updatePendingMerges(merge.maxNumSegmentsOptimize, merge.optimize); + if (success && !merge.isAborted() && !closed && !closing) { + updatePendingMerges(merge.maxNumSegmentsOptimize, merge.optimize, false); + } } } } catch (OutOfMemoryError oom) { @@ -4461,7 +4659,7 @@ merge.abort(); throw new MergePolicy.MergeAbortedException("merge is aborted: " + merge.segString(directory)); } - + final int count = merge.segments.size(); boolean isExternal = false; for(int i=0;i