Index: src/test/org/apache/lucene/index/TestRAMNRT.java =================================================================== --- src/test/org/apache/lucene/index/TestRAMNRT.java (revision 0) +++ src/test/org/apache/lucene/index/TestRAMNRT.java (revision 0) @@ -0,0 +1,205 @@ +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.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Random; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.lucene.analysis.WhitespaceAnalyzer; +import org.apache.lucene.store.Directory; +import org.apache.lucene.store.MockRAMDirectory; +import org.apache.lucene.store.RAMDirectory; +import org.apache.lucene.util.LuceneTestCase; + +public class TestRAMNRT extends LuceneTestCase { + Random random = new Random(); + + public void testSimple() throws Exception { + Directory primaryDir = new MockRAMDirectory(); + RAMDirectory ramDir = new MockRAMDirectory(); + + RAMNRT ramnrt = new RAMNRT(primaryDir, new WhitespaceAnalyzer(), true, + IndexWriter.MaxFieldLength.LIMITED, ramDir, 1024 * 150); // 150k + ramnrt.setInfoStream(System.out); + + IndexWriter primaryWriter = ramnrt.getPrimaryWriter(); + // primaryWriter.setMergeScheduler(new SerialMergeScheduler()); + for (int i = 0; i < 100; i++) { + primaryWriter.addDocument(TestIndexWriterReader.createDocument(i, + "primary", 4)); + } + for (int i = 0; i < 100; i++) { + ramnrt.addDocument(TestIndexWriterReader.createDocument(i, "ram", 4)); + } + + IndexReader reader = ramnrt.getReader(); + assertEquals(200, reader.maxDoc()); + IndexReader ramReader = ramnrt.getRAMWriter().getReader(); + assertEquals(100, ramReader.maxDoc()); + IndexReader primaryReader = ramnrt.getPrimaryWriter().getReader(); + assertEquals(100, primaryReader.maxDoc()); + ramReader.close(); + primaryReader.close(); + reader.close(); + + ramnrt.flush(); + ramReader = ramnrt.getPrimaryWriter().getReader(); + assertEquals(200, ramReader.maxDoc()); + primaryReader = ramnrt.getRAMWriter().getReader(); + assertEquals(0, primaryReader.maxDoc()); + + ramReader.close(); + primaryReader.close(); + + ramnrt.close(); + + primaryDir.close(); + ramDir.close(); + } + + /** + * This tests synchronization of the primary and ram writers. This insures we + * don't get duplicate segment infos from the writers. + * + * @throws Exception + */ + public void testRandomThreads() throws Exception { + long duration = 1000 * 35; + AtomicInteger totalAdded = new AtomicInteger(0); + + Directory primaryDir = new MockRAMDirectory(); + RAMDirectory ramDir = new MockRAMDirectory(); + + RAMNRT ramnrt = new RAMNRT(primaryDir, new WhitespaceAnalyzer(), true, + IndexWriter.MaxFieldLength.LIMITED, ramDir, 1024 * 150); // 150k + ramnrt.setInfoStream(System.out); + + GetReadersThread[] grts = new GetReadersThread[3]; + for (int x = 0; x < grts.length; x++) { + grts[x] = new GetReadersThread("getreadersthread" + x, ramnrt); + } + AddDocsThread[] adts = new AddDocsThread[3]; + for (int x = 0; x < adts.length; x++) { + adts[x] = new AddDocsThread("adddocsthread " + x, ramnrt, totalAdded); + } + for (int x = 0; x < adts.length; x++) { + adts[x].start(); + } + for (int x = 0; x < grts.length; x++) { + grts[x].start(); + } + long startTime = System.currentTimeMillis(); + while (true) { + if ((System.currentTimeMillis() - startTime) > duration) { + break; + } + } + for (int x = 0; x < adts.length; x++) { + adts[x].dorun = false; + } + for (int x = 0; x < grts.length; x++) { + grts[x].dorun = false; + } + for (int x = 0; x < adts.length; x++) { + adts[x].join(); + } + for (int x = 0; x < grts.length; x++) { + grts[x].join(); + } + System.out.println("Completed..."); + IndexReader reader = ramnrt.getReader(); + assertEquals(totalAdded.get(), reader.maxDoc()); + ramnrt.close(); + primaryDir.close(); + ramDir.close(); + } + + public class GenericThread extends Thread { + boolean dorun = true; + + List errors = new ArrayList(); + + RAMNRT ramnrt; + + public GenericThread(String threadName, RAMNRT ramnrt) { + this.ramnrt = ramnrt; + setName(threadName); + } + } + + /** + * Check to insure the readers are not identical + */ + public class GetReadersThread extends GenericThread { + List failures = new ArrayList(); + + public GetReadersThread(String threadName, RAMNRT ramnrt) { + super(threadName, ramnrt); + } + + public void run() { + try { + while (dorun) { + int num = random.nextInt(100); + for (int x=0; x < num; x++) { + IndexReader reader = ramnrt.getReader(); + reader.close(); + } + } + } catch (Throwable ex) { + errors.add(ex); + if (errors.size() > 5) { + throw new RuntimeException(ex); + } + } + } + } + + public class AddDocsThread extends GenericThread { + AtomicInteger totalAdded; + + public AddDocsThread(String threadName, RAMNRT ramnrt, + AtomicInteger totalAdded) { + super(threadName, ramnrt); + this.totalAdded = totalAdded; + } + + public void run() { + try { + while (dorun) { + int numdocs = random.nextInt(2000); + for (int x = 0; x < numdocs; x++) { + if (!dorun) + return; + ramnrt.addDocument(TestIndexWriterReader.createDocument(x, "", 3)); + totalAdded.incrementAndGet(); + } + System.out.println("added " + numdocs + " docs"); + ramnrt.flush(); + } + } catch (Throwable th) { + errors.add(th); + } + } + } +} Index: src/java/org/apache/lucene/index/RAMNRT.java =================================================================== --- src/java/org/apache/lucene/index/RAMNRT.java (revision 0) +++ src/java/org/apache/lucene/index/RAMNRT.java (revision 0) @@ -0,0 +1,343 @@ +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.io.PrintStream; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +import org.apache.lucene.analysis.Analyzer; +import org.apache.lucene.document.Document; +import org.apache.lucene.index.DocumentsWriter.IndexingChain; +import org.apache.lucene.index.IndexWriter.MaxFieldLength; +import org.apache.lucene.search.Query; +import org.apache.lucene.store.Directory; +import org.apache.lucene.store.FileSwitchDirectory; +import org.apache.lucene.store.LockObtainFailedException; +import org.apache.lucene.store.RAMDirectory; + +/** + * Writers that use the same directory. The RAMDir segment names are generated + * from the primary writer. Updates are made to the ramWriter, which is flushed + * when the maximum allocated RAM is exceeded. + */ +// TODO: mergingSegments in ram writer needs to be cleared on abort and exit +// TODO: flush can be performed in the background? +public class RAMNRT { + private 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"); + } + + private RAMIndexWriter ramWriter; + + private PrimaryWriter primaryWriter; + + private RAMDirectory ramDirectory; + + private long maxRam; + + private PrintStream infoStream; + + public RAMNRT(Directory d, Analyzer a, boolean create, MaxFieldLength mfl, + RAMDirectory ramDir, long maxRam) throws IOException { + this.primaryWriter = new PrimaryWriter(d, a, create, mfl); + this.maxRam = maxRam; + Directory primaryDir = primaryWriter.getDirectory(); + ramDirectory = ramDir; + FileSwitchDirectory fsd = new FileSwitchDirectory(SWITCH_FILE_EXTS, + primaryDir, ramDirectory, true); + ramWriter = new RAMIndexWriter(fsd, primaryWriter.getAnalyzer(), true, + new MaxFieldLength(primaryWriter.getMaxFieldLength())); + } + + public void setInfoStream(PrintStream infoStream) { + this.infoStream = infoStream; + } + + public IndexWriter getPrimaryWriter() { + return primaryWriter; + } + + public IndexWriter getRAMWriter() { + return ramWriter; + } + + /** + * Closes the underlying primary and ram writers + * + * @throws IOException + */ + public void close() throws IOException { + ramWriter.close(); + primaryWriter.close(); + } + + /** + * Returns a MultiReader of the readers from the primary and ram writers. + * Duplicates may exist as ram segments are merged to the primary writer, + * this is ok because we remove duplicate here. + * + * @return + * @throws IOException + */ + public IndexReader getReader() throws IOException { + IndexReader[] ramReaders = ramWriter.getReader().getSequentialSubReaders(); + IndexReader[] primaryReaders = primaryWriter.getReader() + .getSequentialSubReaders(); + List readers = new ArrayList(ramReaders.length + primaryReaders.length); + int i = 0; + Set infos = new HashSet(); + for (int x = 0; x < primaryReaders.length; x++) { + SegmentInfo info = ((SegmentReader)primaryReaders[x]).getSegmentInfo(); + infos.add(info); + readers.add(primaryReaders[x]); + } + for (int x = 0; x < ramReaders.length; x++) { + SegmentInfo info = ((SegmentReader)ramReaders[x]).getSegmentInfo(); + if (!infos.contains(info)) { + infos.add(info); + readers.add(ramReaders[x]); + } + } + MultiReader reader = new MultiReader((IndexReader[])readers.toArray(new IndexReader[0])); + return reader; + } + + /** + * Returns the total ram usage, the ram writer's buffer plus the ram + * directory. + */ + public long getRAMSize() { + return ramWriter.ramSizeInBytes() + ramDirectory.sizeInBytes(); + } + + public void addDocument(Document doc) throws IOException { + addDocument(doc, primaryWriter.getAnalyzer()); + } + + public void addDocument(Document doc, Analyzer analyzer) throws IOException { + ramWriter.addDocument(doc, analyzer); + ifFlush(); + } + + public void deleteDocuments(Query query) throws CorruptIndexException, + IOException { + ramWriter.deleteDocuments(query); + primaryWriter.deleteDocuments(query); + ifFlush(); + } + + public void deleteDocuments(Query... queries) throws CorruptIndexException, + IOException { + ramWriter.deleteDocuments(queries); + primaryWriter.deleteDocuments(queries); + } + + public void updateDocument(Term term, Document doc, Analyzer analyzer) + throws CorruptIndexException, IOException { + ramWriter.updateDocument(term, doc, analyzer); + primaryWriter.deleteDocuments(term); + } + + public void updateDocument(Term term, Document doc) + throws CorruptIndexException, IOException { + ramWriter.updateDocument(term, doc); + primaryWriter.deleteDocuments(term); + } + + private void ifFlush() throws IOException { + if (getRAMSize() > maxRam) { + long ramSizeBytes = ramWriter.ramSizeInBytes(); + long ramDirBytes = ramDirectory.sizeInBytes(); + // if (infoStream != null) + // infoStream.println("ramSize:"+ramSizeBytes+" ramDir:"+ramDirBytes); + flush(); + } + } + + /** + * Flushes the RAM writer, and merge all RAM segments to one primary directory + * segment. + * + * @throws IOException + */ + public synchronized void flush() throws IOException { + ramWriter.flush(false, true, true); + // merge all RAM segments to one primary directory segment + SegmentInfos ramInfos = null; + ramInfos = ramWriter.getSegmentInfos(); + MergePolicy.OneMerge merge = new MergePolicy.OneMerge(ramInfos, ramWriter + .getUseCompoundFile()); + merge.ramnrt = true; + synchronized (primaryWriter) { + if (ramWriter.addRAMMerge(merge)) { + primaryWriter.addMerge(merge); + } + } + synchronized (primaryWriter) { + for (SegmentInfo info : ramInfos) { + primaryWriter.segmentInfos.add(info); + } + } + // TODO: needs to be ram nrt segment specific so that + // we don't resolve segments from addIndexes + primaryWriter.resolveExternalSegments(); + } + + public class PrimaryWriter extends IndexWriter { + public PrimaryWriter(Directory d, Analyzer a, boolean create, + IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl, + IndexingChain indexingChain, IndexCommit commit) + throws CorruptIndexException, LockObtainFailedException, IOException { + super(d, a, create, deletionPolicy, mfl, indexingChain, commit); + } + + public PrimaryWriter(Directory d, Analyzer a, boolean create, + IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl) + throws CorruptIndexException, LockObtainFailedException, IOException { + super(d, a, create, deletionPolicy, mfl); + } + + public PrimaryWriter(Directory d, Analyzer a, boolean create, + MaxFieldLength mfl) throws CorruptIndexException, + LockObtainFailedException, IOException { + super(d, a, create, mfl); + } + + public PrimaryWriter(Directory d, Analyzer a, + IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl, + IndexCommit commit) throws CorruptIndexException, + LockObtainFailedException, IOException { + super(d, a, deletionPolicy, mfl, commit); + } + + public PrimaryWriter(Directory d, Analyzer a, + IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl) + throws CorruptIndexException, LockObtainFailedException, IOException { + super(d, a, deletionPolicy, mfl); + } + + public PrimaryWriter(Directory d, Analyzer a, MaxFieldLength mfl) + throws CorruptIndexException, LockObtainFailedException, IOException { + super(d, a, mfl); + } + + void mergeSuccess(MergePolicy.OneMerge merge) { + try { + // here we remove segments from the ram writer that have been merged + // away + if (merge.ramnrt) { + if (infoStream != null) + infoStream.println("primaryWriter.mergeSuccess " + + merge.segString(primaryWriter.getDirectory())); + // it's one of our ram merges + // we need to remove these segments from the ram writer + synchronized (ramWriter) { + System.out.println("ramwriter remove merged segments: " + + merge.segments); + // TODO: assert there are no duplicate segments + ramWriter.removeSegments(merge.segments); + ramWriter.mergeFinish(merge); + ramWriter.removeMerging(merge.segments); + } + } + } catch (IOException ioe) { + // TODO: do something better with this exception + ioe.printStackTrace(); + } + } + } + + public class RAMIndexWriter extends IndexWriter { + Set ramMergingSegments = new HashSet(); + + public RAMIndexWriter(Directory d, Analyzer a, boolean create, + IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl, + IndexingChain indexingChain, IndexCommit commit) + throws CorruptIndexException, LockObtainFailedException, IOException { + super(d, a, create, deletionPolicy, mfl, indexingChain, commit); + } + + public RAMIndexWriter(Directory d, Analyzer a, boolean create, + IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl) + throws CorruptIndexException, LockObtainFailedException, IOException { + super(d, a, create, deletionPolicy, mfl); + } + + public RAMIndexWriter(Directory d, Analyzer a, boolean create, + MaxFieldLength mfl) throws CorruptIndexException, + LockObtainFailedException, IOException { + super(d, a, create, mfl); + } + + public RAMIndexWriter(Directory d, Analyzer a, + IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl, + IndexCommit commit) throws CorruptIndexException, + LockObtainFailedException, IOException { + super(d, a, deletionPolicy, mfl, commit); + } + + public RAMIndexWriter(Directory d, Analyzer a, + IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl) + throws CorruptIndexException, LockObtainFailedException, IOException { + super(d, a, deletionPolicy, mfl); + } + + public RAMIndexWriter(Directory d, Analyzer a, MaxFieldLength mfl) + throws CorruptIndexException, LockObtainFailedException, IOException { + super(d, a, mfl); + } + + public synchronized boolean addRAMMerge(MergePolicy.OneMerge merge) + throws IOException { + int count = merge.segments.size(); + for (int i = 0; i < count; i++) { + final SegmentInfo info = merge.segments.info(i); + if (mergingSegments.contains(info)) + return false; + } + return true; + } + + public void removeMerging(SegmentInfos infos) throws IOException { + mergingSegments.removeAll(infos); + } + + protected synchronized void finishMerges(boolean waitForMerges) throws IOException { + // clear out the somewhat fake ram segment merges + // they're fake because ram writer isn't actually merging them + // however they're marked as being merged so ram writer + // doesn't waste time merging segments that are being + // moved to the primary writer + mergingSegments.removeAll(ramMergingSegments); + super.finishMerges(waitForMerges); + } + + protected String newSegmentName() { + return primaryWriter.newSegmentName(); + } + } +} Index: src/java/org/apache/lucene/index/MergePolicy.java =================================================================== --- src/java/org/apache/lucene/index/MergePolicy.java (revision 831174) +++ src/java/org/apache/lucene/index/MergePolicy.java (working copy) @@ -85,6 +85,7 @@ final boolean useCompoundFile; boolean aborted; Throwable error; + boolean ramnrt = false; public OneMerge(SegmentInfos segments, boolean useCompoundFile) { if (0 == segments.size()) Index: src/java/org/apache/lucene/index/ConcurrentMergeScheduler.java =================================================================== --- src/java/org/apache/lucene/index/ConcurrentMergeScheduler.java (revision 831174) +++ src/java/org/apache/lucene/index/ConcurrentMergeScheduler.java (working copy) @@ -314,6 +314,9 @@ // suppressExceptions is normally only set during // testing. anyExceptions = true; + if (merge.ramnrt) { + System.out.println("ramnrt merge exception"); + } handleMergeException(exc); } } Index: src/java/org/apache/lucene/index/IndexWriter.java =================================================================== --- src/java/org/apache/lucene/index/IndexWriter.java (revision 831174) +++ src/java/org/apache/lucene/index/IndexWriter.java (working copy) @@ -253,7 +253,7 @@ private SegmentInfos localRollbackSegmentInfos; // segmentInfos we will fallback to if the commit fails private int localFlushedDocCount; // saved docWriter.getFlushedDocCount during local transaction - private SegmentInfos segmentInfos = new SegmentInfos(); // the segments + protected SegmentInfos segmentInfos = new SegmentInfos(); // the segments private DocumentsWriter docWriter; private IndexFileDeleter deleter; @@ -269,15 +269,15 @@ // Holds all SegmentInfo instances currently involved in // merges - private HashSet mergingSegments = new HashSet(); + protected HashSet mergingSegments = new HashSet(); private MergePolicy mergePolicy = new LogByteSizeMergePolicy(this); private MergeScheduler mergeScheduler = new ConcurrentMergeScheduler(); - private LinkedList pendingMerges = new LinkedList(); - private Set runningMerges = new HashSet(); + protected LinkedList pendingMerges = new LinkedList(); + protected Set runningMerges = new HashSet(); private List mergeExceptions = new ArrayList(); private long mergeGen; - private boolean stopMerges; + protected boolean stopMerges; private int flushCount; private int flushDeletesCount; @@ -2127,7 +2127,7 @@ return flushDeletesCount; } - final String newSegmentName() { + protected String newSegmentName() { // Cannot synchronize on IndexWriter because that causes // deadlock synchronized(segmentInfos) { @@ -2236,7 +2236,47 @@ public void optimize(boolean doWait) throws CorruptIndexException, IOException { optimize(1, doWait); } - + + /** + * Remove readers and segments . + * @param infos + * @throws CorruptIndexException + * @throws IOException + */ + void removeSegments(SegmentInfos infos) throws CorruptIndexException, IOException { + for (int i=0;i 0; if (stopMerges) @@ -2473,28 +2518,34 @@ } final MergePolicy.MergeSpecification spec; - if (optimize) { - spec = mergePolicy.findMergesForOptimize(segmentInfos, maxNumSegmentsOptimize, segmentsToOptimize); + if (mergeSpec != null) { + spec = mergeSpec; + } else { + if (optimize) { + spec = mergePolicy.findMergesForOptimize(segmentInfos, maxNumSegmentsOptimize, segmentsToOptimize); - if (spec != null) { - final int numMerges = spec.merges.size(); - for(int i=0;i 0) { + MergePolicy.OneMerge externalInfosMerge = new MergePolicy.OneMerge(externalInfos, getUseCompoundFile()); + spec = new MergePolicy.MergeSpecification(); + spec.merges.add(externalInfosMerge); + } + done = true; + if (spec == null) continue; + for(int i=0; i < spec.merges.size(); i++) { + final MergePolicy.OneMerge newMerge = (MergePolicy.OneMerge)spec.merges.get(i); + done = false; + // Returns true if no running merge conflicts + // with this one (and, records this merge as + // pending), ie, this segment is not currently + // being merged: + if (registerMerge(newMerge)) { + merge = newMerge; + + // If this segment is not currently being + // merged, then advance it to running & run + // the merge ourself (below): + pendingMerges.remove(merge); + runningMerges.add(merge); + break; + } + } + + if (!done && merge == null) + // We are not yet done (FSD segments still + // exist in segmentInfos), yet, all such segments + // are currently "covered" by a pending or running + // merge. We now try to grab any pending merge + // that involves FSD segments: + + // we get merges that have ram segments, or are going to ram + // and merge all of them until they're gone + // TODO: ideally we could abort ram -> ram merges however + // they should be fast so it shouldn't matter + merge = getNextRAMNRTMerge(); + + if (!done && merge == null) + // We are not yet done, and, all external segments + // fall under merges that the merge scheduler is + // currently running. So, we now wait and check + // back to see if the merge has completed. + doWait(); + } + + if (merge != null) { + any = true; + merge(merge); + } + } + + if (any) + // Sometimes, on copying an external segment over, + // more merges may become necessary: + mergeScheduler.merge(this); + } + **/ /* If any of our segments are using a directory != ours * then we have to either copy them over one by one, merge * them (if merge policy has chosen to) or wait until @@ -3068,7 +3193,7 @@ * We don't return until the SegmentInfos has no more * external segments. Currently this is only used by * addIndexesNoOptimize(). */ - private void resolveExternalSegments() throws CorruptIndexException, IOException { + void resolveExternalSegments() throws CorruptIndexException, IOException { boolean any = false; @@ -3135,7 +3260,7 @@ // more merges may become necessary: mergeScheduler.merge(this); } - + /** Merges the provided indexes into this index. *

After this completes, the index is optimized.

*

The provided IndexReaders are not closed.

@@ -3667,7 +3792,8 @@ 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); + return -1; + //throw new MergePolicy.MergeException("could not find segment " + merge.segments.info(0).name + " in current index " + segString(), directory); final int numSegments = segmentInfos.size(); @@ -3709,8 +3835,17 @@ // started merging: int docUpto = 0; int delCount = 0; - + int sss = sourceSegments.size(); + if (sss != merge.readersClone.length) { + System.out.println("sss:"+sss+" merge.readersClone.length:"+merge.readersClone.length); + } + if (sss != merge.readers.length) { + System.out.println("sss:"+sss+" merge.readers.length:"+merge.readers.length); + } for(int i=0; i < sourceSegments.size(); i++) { + if (sss != sourceSegments.size()) { + throw new RuntimeException("sss:"+sss+" size: "+sourceSegments.size()); + } SegmentInfo info = sourceSegments.info(i); int docCount = info.docCount; SegmentReader previousReader = merge.readersClone[i]; @@ -3816,9 +3951,14 @@ merge.info.setHasProx(merger.hasProx()); - segmentInfos.subList(start, start + merge.segments.size()).clear(); + //segmentInfos.subList(start, start + merge.segments.size()).clear(); + // remove the segmentinfo's one by one + for (int x=0; x < merge.segments.size(); x++) { + segmentInfos.remove(merge.segments.info(x)); + } assert !segmentInfos.contains(merge.info); - segmentInfos.add(start, merge.info); + if (start >=0) segmentInfos.add(start, merge.info); + else segmentInfos.add(merge.info); // Must note the change to segmentInfos so any commits // in-flight don't lose it: @@ -3917,7 +4057,11 @@ handleOOM(oom, "merge"); } } - + + SegmentInfos getSegmentInfos() { + return segmentInfos; + } + /** Hook that's called when the specified merge is complete. */ void mergeSuccess(MergePolicy.OneMerge merge) { } @@ -3944,13 +4088,13 @@ final SegmentInfo info = merge.segments.info(i); if (mergingSegments.contains(info)) return false; - if (segmentInfos.indexOf(info) == -1) - return false; + //if (segmentInfos.indexOf(info) == -1) + // return false; if (info.dir != directory) isExternal = true; } - ensureContiguousMerge(merge); + //ensureContiguousMerge(merge); pendingMerges.add(merge); @@ -4191,7 +4335,9 @@ message("merging " + merge.segString(directory)); merger = new SegmentMerger(this, mergedName, merge); - + + assert merge.readers == null; + assert merge.readersClone == null; merge.readers = new SegmentReader[numSegments]; merge.readersClone = new SegmentReader[numSegments];