Index: src/test/org/apache/lucene/index/TestPerSegmentDeletes.java =================================================================== --- src/test/org/apache/lucene/index/TestPerSegmentDeletes.java (revision 0) +++ src/test/org/apache/lucene/index/TestPerSegmentDeletes.java (revision 0) @@ -0,0 +1,269 @@ +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.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.lucene.analysis.MockAnalyzer; +import org.apache.lucene.document.Document; +import org.apache.lucene.index.DocumentsWriter.SegmentDeletes; +import org.apache.lucene.index.TermsEnum.SeekStatus; +import org.apache.lucene.store.Directory; +import org.apache.lucene.store.MockDirectoryWrapper; +import org.apache.lucene.store.RAMDirectory; +import org.apache.lucene.util.ArrayUtil; +import org.apache.lucene.util.Bits; +import org.apache.lucene.util.BytesRef; +import org.apache.lucene.util.LuceneTestCase; +import org.apache.lucene.util.Version; + +public class TestPerSegmentDeletes extends LuceneTestCase { + public void testDeletes1() throws Exception { + IndexWriter.debug2 = System.out; + Directory dir = new MockDirectoryWrapper(new RAMDirectory()); + IndexWriterConfig iwc = new IndexWriterConfig(Version.LUCENE_CURRENT, + new MockAnalyzer()); + iwc.setMergeScheduler(new SerialMergeScheduler()); + iwc.setMaxBufferedDocs(5000); + iwc.setRAMBufferSizeMB(400); + RangeMergePolicy fsmp = new RangeMergePolicy(false); + iwc.setMergePolicy(fsmp); + IndexWriter writer = new IndexWriter(dir, iwc); + Document doc = new Document(); + for (int x = 0; x < 5; x++) { + writer.addDocument(TestIndexWriterReader.createDocument(x, "1", 2)); + System.out.println("numRamDocs(" + x + ")" + writer.numRamDocs()); + } + System.out.println("commit1"); + writer.commit(); + //assertEquals(1, writer.docWriter.lastSegmentIndex); + assertEquals(1, writer.segmentInfos.size()); + for (int x = 5; x < 10; x++) { + writer.addDocument(TestIndexWriterReader.createDocument(x, "2", 2)); + System.out.println("numRamDocs(" + x + ")" + writer.numRamDocs()); + } + System.out.println("commit2"); + writer.commit(); + //assertEquals(2, writer.docWriter.lastSegmentIndex); + assertEquals(2, writer.segmentInfos.size()); + + for (int x = 10; x < 15; x++) { + writer.addDocument(TestIndexWriterReader.createDocument(x, "3", 2)); + System.out.println("numRamDocs(" + x + ")" + writer.numRamDocs()); + } + System.out.println("flush(false, false, false)"); + writer.flush(false, false, false); + //assertEquals(2, writer.docWriter.lastSegmentIndex); + assertEquals(3, writer.segmentInfos.size()); + + // added docs are in the ram buffer + for (int x = 15; x < 20; x++) { + writer.addDocument(TestIndexWriterReader.createDocument(x, "4", 2)); + System.out.println("numRamDocs(" + x + ")" + writer.numRamDocs()); + } + assertTrue(writer.numRamDocs() > 0); + // delete from the ram buffer + writer.deleteDocuments(new Term("id", Integer.toString(13))); + + Term id3 = new Term("id", Integer.toString(3)); + + // delete from the 1st segment + writer.deleteDocuments(id3); + + assertTrue(writer.numRamDocs() > 0); + + System.out + .println("segdels1:" + writer.docWriter.deletesToString()); + + assertTrue(writer.docWriter.segmentDeletes.size() > 0); + + // we cause a merge to happen + fsmp.doMerge = true; + fsmp.start = 0; + fsmp.length = 2; + System.out.println("maybeMerge "+writer.segmentInfos); + + SegmentInfo info0 = writer.segmentInfos.get(0); + SegmentInfo info1 = writer.segmentInfos.get(1); + + writer.maybeMerge(); + System.out.println("maybeMerge after "+writer.segmentInfos); + // there should be docs in RAM + assertTrue(writer.numRamDocs() > 0); + + // assert we've merged the 1 and 2 segments + // and still have a segment leftover == 2 + assertEquals(2, writer.segmentInfos.size()); + assertFalse(segThere(info0, writer.segmentInfos)); + assertFalse(segThere(info1, writer.segmentInfos)); + + System.out.println("segdels2:" + writer.docWriter.deletesToString()); + + assertTrue(writer.docWriter.segmentDeletes.size() > 0); + + IndexReader r = writer.getReader(); + IndexReader r1 = r.getSequentialSubReaders()[0]; + printDelDocs(r1.getDeletedDocs()); + int[] docs = toDocsArray(id3, null, r); + System.out.println("id3 docs:"+Arrays.toString(docs)); + // there shouldn't be any docs for id:3 + assertTrue(docs == null); + r.close(); + + part2(writer, fsmp); + + // System.out.println("segdels2:"+writer.docWriter.segmentDeletes.toString()); + System.out.println("close"); + writer.close(); + dir.close(); + } + + void part2(IndexWriter writer, RangeMergePolicy fsmp) throws Exception { + for (int x = 20; x < 25; x++) { + writer.addDocument(TestIndexWriterReader.createDocument(x, "5", 2)); + System.out.println("numRamDocs(" + x + ")" + writer.numRamDocs()); + } + writer.flush(false, true, false); + for (int x = 25; x < 30; x++) { + writer.addDocument(TestIndexWriterReader.createDocument(x, "5", 2)); + System.out.println("numRamDocs(" + x + ")" + writer.numRamDocs()); + } + writer.flush(false, true, false); + + System.out.println("infos3:"+writer.segmentInfos); + + Term delterm = new Term("id", "8"); + writer.deleteDocuments(delterm); + System.out.println("segdels3:" + writer.docWriter.deletesToString()); + + fsmp.doMerge = true; + fsmp.start = 1; + fsmp.length = 2; + writer.maybeMerge(); + + // deletes for info1, the newly created segment from the + // merge should have no deletes because they were applied in + // the merge + SegmentInfo info1 = writer.segmentInfos.get(1); + assertFalse(exists(info1, writer.docWriter.segmentDeletes)); + + System.out.println("infos4:"+writer.segmentInfos); + System.out.println("segdels4:" + writer.docWriter.deletesToString()); + } + + static boolean exists(SegmentInfo info, Map map) { + if (!map.containsKey(info)) return false; + SegmentDeletes dels = map.get(info); + return dels.any(); + } + + boolean segThere(SegmentInfo info, SegmentInfos infos) { + for (SegmentInfo si : infos) { + if (si.name.equals(info.name)) return true; + } + return false; + } + + public static void printDelDocs(Bits bits) { + if (bits == null) return; + for (int x = 0; x < bits.length(); x++) { + System.out.println(x + ":" + bits.get(x)); + } + } + + public static int[] toDocsArray(Term term, Bits bits, IndexReader reader) + throws IOException { + Fields fields = MultiFields.getFields(reader); + Terms cterms = fields.terms(term.field); + TermsEnum ctermsEnum = cterms.iterator(); + SeekStatus ss = ctermsEnum.seek(new BytesRef(term.text()), false); + if (ss.equals(SeekStatus.FOUND)) { + DocsEnum docsEnum = ctermsEnum.docs(bits, null); + return toArray(docsEnum); + } + return null; + } + + public static int[] toArray(DocsEnum docsEnum) throws IOException { + List docs = new ArrayList(); + while (docsEnum.nextDoc() != DocsEnum.NO_MORE_DOCS) { + int docID = docsEnum.docID(); + docs.add(docID); + } + return ArrayUtil.toIntArray(docs); + } + + public class RangeMergePolicy extends MergePolicy { + boolean doMerge = false; + int start; + int length; + + private final boolean useCompoundFile; + + private RangeMergePolicy(boolean useCompoundFile) { + this.useCompoundFile = useCompoundFile; + } + + @Override + public void close() {} + + public MergeSpecification findMerges(SegmentInfos segmentInfos) + throws CorruptIndexException, IOException { + MergeSpecification ms = new MergeSpecification(); + if (doMerge) { + SegmentInfos mergeInfos = new SegmentInfos(); + for (int x=start; x < (start+length); x++) { + mergeInfos.add(segmentInfos.get(x)); + } + OneMerge om = new OneMerge(mergeInfos, false); + ms.add(om); + doMerge = false; + return ms; + } + return null; + } + + @Override + public MergeSpecification findMergesForOptimize(SegmentInfos segmentInfos, + int maxSegmentCount, Set segmentsToOptimize) + throws CorruptIndexException, IOException { + return null; + } + + @Override + public MergeSpecification findMergesToExpungeDeletes( + SegmentInfos segmentInfos) throws CorruptIndexException, IOException { + return null; + } + + @Override + public boolean useCompoundDocStore(SegmentInfos segments) { + return useCompoundFile; + } + + @Override + public boolean useCompoundFile(SegmentInfos segments, SegmentInfo newSegment) { + return useCompoundFile; + } + } +} Index: src/java/org/apache/lucene/index/BufferedDeletes.java =================================================================== --- src/java/org/apache/lucene/index/BufferedDeletes.java (revision 1030113) +++ src/java/org/apache/lucene/index/BufferedDeletes.java (working copy) @@ -49,6 +49,13 @@ terms = new HashMap(); } } + + @Override + public String toString() { + return "BufferedDeletes [numTerms=" + numTerms + ", terms=" + terms + + ", queries=" + queries + ", docIDs=" + docIDs + ", bytesUsed=" + + bytesUsed + ", doTermSort=" + doTermSort + "]"; + } // Number of documents a delete term applies to. final static class Num { Index: src/java/org/apache/lucene/index/MergePolicy.java =================================================================== --- src/java/org/apache/lucene/index/MergePolicy.java (revision 1030113) +++ src/java/org/apache/lucene/index/MergePolicy.java (working copy) @@ -17,15 +17,17 @@ * limitations under the License. */ +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; + +import org.apache.lucene.index.DocumentsWriter.SegmentDeletes; +import org.apache.lucene.index.MergePolicy.MergeSpecification; import org.apache.lucene.store.Directory; import org.apache.lucene.util.SetOnce; import org.apache.lucene.util.SetOnce.AlreadySetException; -import java.io.IOException; -import java.util.List; -import java.util.ArrayList; -import java.util.Set; - /** *

Expert: a MergePolicy determines the sequence of * primitive merge operations to be used for overall merge Index: src/java/org/apache/lucene/index/DocumentsWriter.java =================================================================== --- src/java/org/apache/lucene/index/DocumentsWriter.java (revision 1030113) +++ src/java/org/apache/lucene/index/DocumentsWriter.java (working copy) @@ -17,18 +17,22 @@ * limitations under the License. */ +import static org.apache.lucene.util.ByteBlockPool.BYTE_BLOCK_MASK; +import static org.apache.lucene.util.ByteBlockPool.BYTE_BLOCK_SIZE; + import java.io.IOException; import java.io.PrintStream; import java.text.NumberFormat; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.HashMap; -import java.util.Map; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Map.Entry; +import java.util.Set; import java.util.concurrent.atomic.AtomicLong; -import java.util.concurrent.locks.ReentrantLock; import org.apache.lucene.analysis.Analyzer; import org.apache.lucene.document.Document; @@ -43,11 +47,9 @@ import org.apache.lucene.store.RAMFile; import org.apache.lucene.util.ArrayUtil; import org.apache.lucene.util.Constants; +import org.apache.lucene.util.RamUsageEstimator; import org.apache.lucene.util.RecyclingByteBlockAllocator; import org.apache.lucene.util.ThreadInterruptedException; -import org.apache.lucene.util.RamUsageEstimator; -import static org.apache.lucene.util.ByteBlockPool.BYTE_BLOCK_MASK; -import static org.apache.lucene.util.ByteBlockPool.BYTE_BLOCK_SIZE; /** * This class accepts multiple added documents and directly @@ -152,7 +154,101 @@ private final int maxThreadStates; List newFiles; + Map segmentDeletes = new HashMap(); + int lastSegmentIndex = 0; + SegmentInfo lastSegment; + final AtomicLong segmentDeletesBytesUsed = new AtomicLong(0); + private static Term[] EMPTY_TERM_ARRAY = new Term[0]; + + public String deletesToString() { + return "lastSegmentIndex:"+lastSegmentIndex+" segmentDeletes:"+segmentDeletes+" deletesInRAM:"+deletesInRAM+" deletesFlushed:"+deletesFlushed; + } + + // if the index is less than or equals to the last segment index + // return the deletes of the last segment + synchronized SegmentDeletes getSegmentDeletes(int index, SegmentInfos allInfos) { + assert index >= 0; + + SegmentInfo info = null; + if (index <= lastSegmentIndex) { + info = allInfos.get(lastSegmentIndex); + } else { + info = allInfos.get(index); + } + return segmentDeletes.get(info); + } + + static class SegmentDeletes { + private Set terms = new HashSet(); + private Set queries = new HashSet(); + SegmentInfo info; + long bytesUsed = 0; + + public SegmentDeletes(SegmentInfo info) { + this.info = info; + } + + void merge(SegmentDeletes mergeDels) { + assert mergeDels.info.equals(info); + addTerms(mergeDels.terms); + addQueries(mergeDels.queries); + } + + void addTerm(Term term) { + if (terms.add(term)) { + // we're only adding the term pointer, not the actual + // term bytes because those have been added for + // deletes flushed + bytesUsed += BYTES_PER_DEL_TERM; + } + } + + void addTerms(Collection ts) { + for (Term t : ts) { + addTerm(t); + } + } + + void addQueries(Collection qs) { + for (Query q : qs) { + addQuery(q); + } + } + + void addQuery(Query query) { + if (queries.add(query)) { + bytesUsed += BYTES_PER_DEL_QUERY; + } + } + + public void clear(DocumentsWriter dw) { + // remove the bytes used by this deletes object + for (int x=0; x < terms.size(); x++) { + dw.segmentDeletesBytesUsed.addAndGet(-BYTES_PER_DEL_TERM); + } + for (int x=0; x < queries.size(); x++) { + dw.segmentDeletesBytesUsed.addAndGet(-BYTES_PER_DEL_QUERY); + } + terms.clear(); + queries.clear(); + } + + public Term[] getSortedTerms() { + Term[] arr = terms.toArray(EMPTY_TERM_ARRAY); + Arrays.sort(arr); + return arr; + } + + @Override + public String toString() { + return "Deletes [terms=" + terms + ", queries=" + queries + "]"; + } + boolean any() { + return terms.size() > 0 || queries.size() > 0; + } + } + static class DocState { DocumentsWriter docWriter; Analyzer analyzer; @@ -329,6 +425,9 @@ if (consumer instanceof DocFieldProcessor) { docFieldProcessor = (DocFieldProcessor) consumer; } + if (writer.segmentInfos.size() > 0) { + lastSegmentIndex = writer.segmentInfos.size() - 1; + } } /** Returns true if any of the fields in the current @@ -735,7 +834,7 @@ * flush is pending. If delTerm is non-null then we * buffer this deleted term after the thread state has * been acquired. */ - synchronized DocumentsWriterThreadState getThreadState(Document doc, Term delTerm) throws IOException { + synchronized DocumentsWriterThreadState getThreadState(Document doc, Term delTerm, SegmentInfos infos) throws IOException { final Thread currentThread = Thread.currentThread(); @@ -785,7 +884,7 @@ assert writer.testPoint("DocumentsWriter.ThreadState.init start"); if (delTerm != null) { - addDeleteTerm(delTerm, state.docState.docID); + addDeleteTerm(delTerm, state.docState.docID, infos); state.doFlushAfter = timeToFlushDeletes(); } @@ -822,21 +921,21 @@ /** Returns true if the caller (IndexWriter) should now * flush. */ - boolean addDocument(Document doc, Analyzer analyzer) + boolean addDocument(Document doc, Analyzer analyzer, SegmentInfos infos) throws CorruptIndexException, IOException { - return updateDocument(doc, analyzer, null); + return updateDocument(doc, analyzer, null, infos); } - boolean updateDocument(Term t, Document doc, Analyzer analyzer) + boolean updateDocument(Term t, Document doc, Analyzer analyzer, SegmentInfos infos) throws CorruptIndexException, IOException { - return updateDocument(doc, analyzer, t); + return updateDocument(doc, analyzer, t, infos); } - boolean updateDocument(Document doc, Analyzer analyzer, Term delTerm) + boolean updateDocument(Document doc, Analyzer analyzer, Term delTerm, SegmentInfos infos) throws CorruptIndexException, IOException { // This call is synchronized but fast - final DocumentsWriterThreadState state = getThreadState(doc, delTerm); + final DocumentsWriterThreadState state = getThreadState(doc, delTerm, infos); final DocState docState = state.docState; docState.doc = doc; @@ -939,43 +1038,43 @@ throw new AlreadyClosedException("this IndexWriter is closed"); } - boolean bufferDeleteTerms(Term[] terms) throws IOException { + boolean bufferDeleteTerms(Term[] terms, SegmentInfos infos) throws IOException { synchronized(this) { waitReady(null); for (int i = 0; i < terms.length; i++) - addDeleteTerm(terms[i], numDocsInRAM); + addDeleteTerm(terms[i], numDocsInRAM, infos); } return timeToFlushDeletes(); } - boolean bufferDeleteTerm(Term term) throws IOException { + boolean bufferDeleteTerm(Term term, SegmentInfos infos) throws IOException { synchronized(this) { waitReady(null); - addDeleteTerm(term, numDocsInRAM); + addDeleteTerm(term, numDocsInRAM, infos); } return timeToFlushDeletes(); } - boolean bufferDeleteQueries(Query[] queries) throws IOException { + boolean bufferDeleteQueries(Query[] queries, SegmentInfos infos) throws IOException { synchronized(this) { waitReady(null); for (int i = 0; i < queries.length; i++) - addDeleteQuery(queries[i], numDocsInRAM); + addDeleteQuery(queries[i], numDocsInRAM, infos); } return timeToFlushDeletes(); } - boolean bufferDeleteQuery(Query query) throws IOException { + boolean bufferDeleteQuery(Query query, SegmentInfos infos) throws IOException { synchronized(this) { waitReady(null); - addDeleteQuery(query, numDocsInRAM); + addDeleteQuery(query, numDocsInRAM, infos); } return timeToFlushDeletes(); } synchronized boolean deletesFull() { return (ramBufferSize != IndexWriterConfig.DISABLE_AUTO_FLUSH && - (deletesInRAM.bytesUsed + deletesFlushed.bytesUsed + bytesUsed()) >= ramBufferSize) || + (segmentDeletesBytesUsed.get()+deletesInRAM.bytesUsed + deletesFlushed.bytesUsed + bytesUsed()) >= ramBufferSize) || (maxBufferedDeleteTerms != IndexWriterConfig.DISABLE_AUTO_FLUSH && ((deletesInRAM.size() + deletesFlushed.size()) >= maxBufferedDeleteTerms)); } @@ -990,7 +1089,7 @@ // when merges (which always apply deletes) are // infrequent. return (ramBufferSize != IndexWriterConfig.DISABLE_AUTO_FLUSH && - (deletesInRAM.bytesUsed + deletesFlushed.bytesUsed) >= ramBufferSize/2) || + (segmentDeletesBytesUsed.get()+deletesInRAM.bytesUsed + deletesFlushed.bytesUsed) >= ramBufferSize/2) || (maxBufferedDeleteTerms != IndexWriterConfig.DISABLE_AUTO_FLUSH && ((deletesInRAM.size() + deletesFlushed.size()) >= maxBufferedDeleteTerms)); } @@ -1009,13 +1108,97 @@ int getMaxBufferedDeleteTerms() { return maxBufferedDeleteTerms; } - + synchronized boolean hasDeletes() { - return deletesFlushed.any(); + if (deletesFlushed.any()) return true; + for (SegmentDeletes dels : segmentDeletes.values()) { + if (dels.any()) { + return true; + } + } + return false; } + + /** + * Apply deletes from the segment deletes map to individual readers. + */ + synchronized boolean applyDeletesToSegments(SegmentInfos allInfos, int start, int length, MergePolicy.OneMerge merge) throws IOException { + IndexWriter.debug2("applyDeletesToSegments start:"+start+" length:"+length); + if (!hasDeletes()) + return false; + + SegmentDeletes coalescedDeletes = null; + if (start > 0) { + coalescedDeletes = coalesceDeletes(allInfos, start, length); + } + + final long t0 = System.currentTimeMillis(); - synchronized boolean applyDeletes(SegmentInfos infos) throws IOException { + if (infoStream != null) + message("apply " + deletesFlushed.numTerms + " buffered deleted terms and " + + deletesFlushed.docIDs.size() + " deleted docIDs and " + + deletesFlushed.queries.size() + " deleted queries on " + + + length + " segments."); + boolean any = false; + + for (int i = start; i < (length+start); i++) { + SegmentInfo info = allInfos.info(i); + // Make sure we never attempt to apply deletes to + // segment in external dir + assert info.dir == directory; + + SegmentDeletes deletes = getSegmentDeletes(i, allInfos); + + if (deletes != null) { + SegmentReader reader = writer.readerPool.get(info, false); + try { + any |= applyDeletesToSegment(deletes, reader); + + // the deletes that have been applied + // remove them from the segment deletes map + deletes.clear(this); + segmentDeletes.remove(info); + } finally { + writer.readerPool.release(reader); + } + } + } + if (infoStream != null) { + message("apply deletes took " + (System.currentTimeMillis()-t0) + " msec"); + } + if (coalescedDeletes != null) { + assert start > 0; + SegmentInfo info = allInfos.get(start-1); + segmentDeletes.put(info, coalescedDeletes); + } + return any; + } + + /** + * If there's a (start - 1) segment, we merge deletes + * from the segments up to (start + length). + */ + synchronized SegmentDeletes coalesceDeletes(SegmentInfos allInfos, int start, int length) { + if (start > 0) { + int mergeSegIndex = start - 1; + SegmentInfo mergeSegInfo = allInfos.get(mergeSegIndex); + SegmentDeletes mergedDeletes = new SegmentDeletes(mergeSegInfo); + for (int x=lastSegmentIndex; x < (start+length); x++) { + SegmentInfo info = allInfos.get(x); + SegmentDeletes d = segmentDeletes.get(info); + if (d != null) { + mergedDeletes.addTerms(d.terms); + mergedDeletes.addQueries(d.queries); + } + } + return mergedDeletes; + } + return null; + } + + synchronized boolean applyDeletesAll(SegmentInfos infos, int newLastSegmentIndex) throws IOException { + if (!hasDeletes()) return false; @@ -1039,7 +1222,7 @@ SegmentReader reader = writer.readerPool.get(infos.info(i), false); try { - any |= applyDeletes(reader, docStart); + any |= applyDeletesAllPerReader(reader, docStart); docStart += reader.maxDoc(); } finally { writer.readerPool.release(reader); @@ -1047,13 +1230,20 @@ } deletesFlushed.clear(); + if (infoStream != null) { message("apply deletes took " + (System.currentTimeMillis()-t0) + " msec"); } - + // because all pending deletes have now been applied + // clear out the pending per-segment deletes + for (SegmentDeletes d : segmentDeletes.values()) { + d.clear(this); + } + segmentDeletes.clear(); + this.lastSegmentIndex = newLastSegmentIndex; return any; } - + // used only by assert private Term lastDeleteTerm; @@ -1065,10 +1255,88 @@ lastDeleteTerm = term; return true; } + + private final synchronized boolean applyDeletesToSegment(SegmentDeletes deletes, SegmentReader reader) throws CorruptIndexException, IOException { + IndexWriter.debug2("applyDeletesToSegment seg:"+reader.getSegmentName()); + boolean any = false; + assert checkDeleteTerm(null); + + // Delete by term + if (deletes.terms.size() > 0) { + Fields fields = reader.fields(); + if (fields == null) { + // This reader has no postings + return false; + } + + TermsEnum termsEnum = null; + + String currentField = null; + DocsEnum docs = null; + + for (Term term : deletes.getSortedTerms()) { + // Since we visit terms sorted, we gain performance + // by re-using the same TermsEnum and seeking only + // forwards + if (term.field() != currentField) { + assert currentField == null || currentField.compareTo(term.field()) < 0; + currentField = term.field(); + Terms terms = fields.terms(currentField); + if (terms != null) { + termsEnum = terms.iterator(); + } else { + termsEnum = null; + } + } + + if (termsEnum == null) { + continue; + } + assert checkDeleteTerm(term); + + if (termsEnum.seek(term.bytes(), false) == TermsEnum.SeekStatus.FOUND) { + DocsEnum docsEnum = termsEnum.docs(reader.getDeletedDocs(), docs); + + if (docsEnum != null) { + docs = docsEnum; + while (true) { + final int docID = docs.nextDoc(); + if (docID == DocsEnum.NO_MORE_DOCS) break; + reader.deleteDocument(docID); + any = true; + } + } + } + } + } + + // Delete by query + if (deletes.queries.size() > 0) { + IndexSearcher searcher = new IndexSearcher(reader); + try { + for (Query query : deletes.queries) { + Weight weight = query.weight(searcher); + Scorer scorer = weight.scorer(reader, true, false); + if (scorer != null) { + while(true) { + int doc = scorer.nextDoc(); + if (doc == DocsEnum.NO_MORE_DOCS) break; + reader.deleteDocument(doc); + any = true; + } + } + } + } finally { + searcher.close(); + } + } + return any; + } + // Apply buffered delete terms, queries and docIDs to the // provided reader - private final synchronized boolean applyDeletes(IndexReader reader, int docIDStart) + private final synchronized boolean applyDeletesAllPerReader(IndexReader reader, int docIDStart) throws CorruptIndexException, IOException { final int docEnd = docIDStart + reader.maxDoc(); @@ -1168,7 +1436,7 @@ // current number of documents buffered in ram so that the // delete term will be applied to those documents as well // as the disk segments. - synchronized private void addDeleteTerm(Term term, int docCount) { + synchronized private void addDeleteTerm(Term term, int docCount, SegmentInfos infos) { BufferedDeletes.Num num = deletesInRAM.terms.get(term); final int docIDUpto = flushedDocCount + docCount; if (num == null) @@ -1178,7 +1446,24 @@ deletesInRAM.numTerms++; deletesInRAM.addBytesUsed(BYTES_PER_DEL_TERM + term.bytes.length); + + assert lastSegmentIndex >= 0; + + // add the delete term to the segments after lastSegmentIndex + for (int x=lastSegmentIndex; x < infos.size(); x++) { + SegmentInfo info = infos.get(x); + addDeleteTermToSegment(info, term); + } } + + synchronized void addDeleteTermToSegment(SegmentInfo info, Term term) { + SegmentDeletes dels = segmentDeletes.get(info); + if (dels == null) { + dels = new SegmentDeletes(info); + segmentDeletes.put(info, dels); + } + dels.addTerm(term); + } // Buffer a specific docID for deletion. Currently only // used when we hit a exception when adding a document @@ -1187,9 +1472,22 @@ deletesInRAM.addBytesUsed(BYTES_PER_DEL_DOCID); } - synchronized private void addDeleteQuery(Query query, int docID) { + synchronized private void addDeleteQuery(Query query, int docID, SegmentInfos infos) { deletesInRAM.queries.put(query, Integer.valueOf(flushedDocCount + docID)); deletesInRAM.addBytesUsed(BYTES_PER_DEL_QUERY); + + assert lastSegmentIndex >= 0; + + // add the delete query to the segments after lastSegmentIndex + for (int x=lastSegmentIndex; x < infos.size(); x++) { + SegmentInfo info = infos.get(x); + SegmentDeletes dels = segmentDeletes.get(info); + if (dels == null) { + dels = new SegmentDeletes(info); + segmentDeletes.put(info, dels); + } + dels.queries.add(query); + } } /** Does the synchronized work to finish/flush the @@ -1373,7 +1671,7 @@ return; } - deletesRAMUsed = deletesInRAM.bytesUsed+deletesFlushed.bytesUsed; + deletesRAMUsed = deletesInRAM.bytesUsed+deletesFlushed.bytesUsed+segmentDeletesBytesUsed.get(); doBalance = bytesUsed() +deletesRAMUsed >= ramBufferSize; } Index: src/java/org/apache/lucene/index/IndexWriter.java =================================================================== --- src/java/org/apache/lucene/index/IndexWriter.java (revision 1030113) +++ src/java/org/apache/lucene/index/IndexWriter.java (working copy) @@ -30,6 +30,7 @@ import org.apache.lucene.store.BufferedIndexInput; import org.apache.lucene.util.Constants; import org.apache.lucene.index.codecs.CodecProvider; +import org.apache.lucene.index.DocumentsWriter.SegmentDeletes; import org.apache.lucene.util.ThreadInterruptedException; import org.apache.lucene.util.Version; import org.apache.lucene.util.Bits; @@ -279,9 +280,9 @@ volatile SegmentInfos pendingCommit; // set when a commit is pending (after prepareCommit() & before commit()) volatile long pendingCommitChangeCount; - private SegmentInfos segmentInfos = new SegmentInfos(); // the segments + 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 @@ -329,7 +330,14 @@ // The PayloadProcessorProvider to use when segments are merged private PayloadProcessorProvider payloadProcessorProvider; - + + static PrintStream debug2 = null; + static void debug2(String msg) { + if (debug2 != null) { + debug2.println(msg); + } + } + /** * Expert: returns a readonly reader, covering all * committed as well as un-committed changes to the index. @@ -408,7 +416,7 @@ // reader; in theory we could do similar retry logic, // just like we do when loading segments_N synchronized(this) { - applyDeletes(); + applyDeletesAll(); final IndexReader r = new DirectoryReader(this, segmentInfos, config.getReaderTermsIndexDivisor(), codecs); if (infoStream != null) { message("return reader version=" + r.getVersion() + " reader=" + r); @@ -1106,13 +1114,11 @@ message("init: loaded commit \"" + commit.getSegmentsFileName() + "\""); } } - setRollbackSegmentInfos(segmentInfos); docWriter = new DocumentsWriter(directory, this, conf.getIndexingChain(), conf.getMaxThreadStates()); docWriter.setInfoStream(infoStream); docWriter.setMaxFieldLength(maxFieldLength); - // Default deleter (for backwards compatibility) is // KeepOnlyLastCommitDeleter: deleter = new IndexFileDeleter(directory, @@ -2001,7 +2007,19 @@ public void addDocument(Document doc) throws CorruptIndexException, IOException { addDocument(doc, analyzer); } - + + synchronized SegmentInfos getSegmentInfos(boolean copy) { + if (copy) { + SegmentInfos copiedInfos = new SegmentInfos(); + for (SegmentInfo info : segmentInfos) { + copiedInfos.add(info); + } + return copiedInfos; + } else { + return segmentInfos; + } + } + /** * Adds a document to this index, using the provided analyzer instead of the * value of {@link #getAnalyzer()}. If the document contains more than @@ -2025,7 +2043,7 @@ boolean success = false; try { try { - doFlush = docWriter.addDocument(doc, analyzer); + doFlush = docWriter.addDocument(doc, analyzer, getSegmentInfos(true)); success = true; } finally { if (!success) { @@ -2065,7 +2083,7 @@ public void deleteDocuments(Term term) throws CorruptIndexException, IOException { ensureOpen(); try { - boolean doFlush = docWriter.bufferDeleteTerm(term); + boolean doFlush = docWriter.bufferDeleteTerm(term, getSegmentInfos(true)); if (doFlush) flush(true, false, false); } catch (OutOfMemoryError oom) { @@ -2089,7 +2107,7 @@ public void deleteDocuments(Term... terms) throws CorruptIndexException, IOException { ensureOpen(); try { - boolean doFlush = docWriter.bufferDeleteTerms(terms); + boolean doFlush = docWriter.bufferDeleteTerms(terms, getSegmentInfos(true)); if (doFlush) flush(true, false, false); } catch (OutOfMemoryError oom) { @@ -2110,7 +2128,7 @@ */ public void deleteDocuments(Query query) throws CorruptIndexException, IOException { ensureOpen(); - boolean doFlush = docWriter.bufferDeleteQuery(query); + boolean doFlush = docWriter.bufferDeleteQuery(query, getSegmentInfos(true)); if (doFlush) flush(true, false, false); } @@ -2130,7 +2148,7 @@ */ public void deleteDocuments(Query... queries) throws CorruptIndexException, IOException { ensureOpen(); - boolean doFlush = docWriter.bufferDeleteQueries(queries); + boolean doFlush = docWriter.bufferDeleteQueries(queries, getSegmentInfos(true)); if (doFlush) flush(true, false, false); } @@ -2182,7 +2200,7 @@ boolean doFlush = false; boolean success = false; try { - doFlush = docWriter.updateDocument(term, doc, analyzer); + doFlush = docWriter.updateDocument(term, doc, analyzer, getSegmentInfos(true)); success = true; } finally { if (!success) { @@ -3401,7 +3419,7 @@ } if (flushDeletes) { - applyDeletes(); + applyDeletesAll(); } if (flushDocs) @@ -3569,14 +3587,14 @@ commitMergedDeletes(merge, mergedReader); docWriter.remapDeletes(segmentInfos, merger.getDocMaps(), merger.getDelCounts(), merge, mergedDocCount); - + setMergeDocStoreIsCompoundFile(merge); merge.info.setHasProx(merger.hasProx()); - + segmentInfos.subList(start, start + merge.segments.size()).clear(); assert !segmentInfos.contains(merge.info); segmentInfos.add(start, merge.info); - + // Must note the change to segmentInfos so any commits // in-flight don't lose it: checkpoint(); @@ -3683,6 +3701,7 @@ /** Hook that's called when the specified merge is complete. */ void mergeSuccess(MergePolicy.OneMerge merge) { + } /** Checks whether this merge involves any segments @@ -3767,11 +3786,14 @@ if (merge.isAborted()) return; - applyDeletes(); - final SegmentInfos sourceSegments = merge.segments; + + int start = ensureContiguousMerge(merge); + final int end = sourceSegments.size(); - + + docWriter.applyDeletesToSegments(segmentInfos, start, end, merge); + // Check whether this merge will allow us to skip // merging the doc stores (stored field & vectors). // This is a very substantial optimization (saves tons @@ -3935,7 +3957,7 @@ mergingSegments.remove(merge.info); merge.registerDone = false; } - + runningMerges.remove(merge); } @@ -4225,7 +4247,7 @@ } // Apply buffered deletes to all segments. - private final synchronized boolean applyDeletes() throws CorruptIndexException, IOException { + private final synchronized boolean applyDeletesAll() throws CorruptIndexException, IOException { assert testPoint("startApplyDeletes"); if (infoStream != null) { message("applyDeletes"); @@ -4234,7 +4256,7 @@ boolean success = false; boolean changed; try { - changed = docWriter.applyDeletes(segmentInfos); + changed = docWriter.applyDeletesAll(segmentInfos, segmentInfos.size()-1); success = true; } finally { if (!success && infoStream != null) {