= dataLen) {
+ dataLen = input.read(ioBuffer);
+ bufferIndex = 0;
+ }
+
+ if (dataLen == -1) {
+ if (length > 0)
+ break;
+ else
+ return null;
+ } else
+ c = ioBuffer[bufferIndex++];
+
+ if (c != ' ') { // if it's a token char
+
+ if (length == 0) // start of token
+ start = offset - 1;
+
+ buffer[length++] = c;
+
+ if (length == MAX_WORD_LEN) // buffer overflow!
+ break;
+
+ } else if (length > 0) // at non-Letter w/ chars
+ break; // return 'em
+ }
+
+ // t.termText = new String(buffer, 0, length);
+ t.termBufferLength = length;
+ t.startOffset = start;
+ t.endOffset = start+length;
+
+ return t;
+ }
+}
+
Property changes on: src/java/org/apache/lucene/analysis/SimpleSpaceTokenizer.java
___________________________________________________________________
Name: svn:eol-style
+ native
Index: src/java/org/apache/lucene/analysis/SimpleSpaceAnalyzer.java
===================================================================
--- src/java/org/apache/lucene/analysis/SimpleSpaceAnalyzer.java (revision 0)
+++ src/java/org/apache/lucene/analysis/SimpleSpaceAnalyzer.java (revision 0)
@@ -0,0 +1,35 @@
+package org.apache.lucene.analysis;
+
+/**
+ * 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.Reader;
+
+/** An Analyzer that uses SimpleSpaceTokenizer. */
+
+public final class SimpleSpaceAnalyzer extends Analyzer {
+ private ThreadLocal tokenizers = new ThreadLocal();
+ public TokenStream tokenStream(String fieldName, Reader reader) {
+ SimpleSpaceTokenizer s = (SimpleSpaceTokenizer) tokenizers.get();
+ if (s == null) {
+ s = new SimpleSpaceTokenizer();
+ tokenizers.set(s);
+ }
+ s.init(reader);
+ return s;
+ }
+}
Property changes on: src/java/org/apache/lucene/analysis/SimpleSpaceAnalyzer.java
___________________________________________________________________
Name: svn:eol-style
+ native
Index: src/java/org/apache/lucene/analysis/Token.java
===================================================================
--- src/java/org/apache/lucene/analysis/Token.java (revision 533617)
+++ src/java/org/apache/lucene/analysis/Token.java (working copy)
@@ -56,7 +56,13 @@
String type = "word"; // lexical type
Payload payload;
-
+
+ // For better indexing performance, set buffer & length
+ // instead of termText
+ char[] termBuffer;
+ int termBufferOffset;
+ int termBufferLength;
+
private int positionIncrement = 1;
/** Constructs a Token with the given term text, and start & end offsets.
@@ -67,6 +73,17 @@
endOffset = end;
}
+ /** Constructs a Token with the given term text buffer
+ starting at offset for length lenth, and start & end offsets.
+ The type defaults to "word." */
+ public Token(char[] text, int offset, int length, int start, int end) {
+ termBuffer = text;
+ termBufferOffset = offset;
+ termBufferLength = length;
+ startOffset = start;
+ endOffset = end;
+ }
+
/** Constructs a Token with the given text, start and end offsets, & type. */
public Token(String text, int start, int end, String typ) {
termText = text;
@@ -75,6 +92,19 @@
type = typ;
}
+ /** Constructs a Token with the given term text buffer
+ starting at offset for length lenth, and start & end
+ offsets, & type. */
+ public Token(char[] text, int offset, int length, int start, int end, String typ) {
+ termBuffer = text;
+ termBufferOffset = offset;
+ termBufferLength = length;
+ startOffset = start;
+ endOffset = end;
+ type = typ;
+ }
+
+
/** Set the position increment. This determines the position of this token
* relative to the previous Token in a {@link TokenStream}, used in phrase
* searching.
@@ -119,7 +149,20 @@
/** Returns the Token's term text. */
public final String termText() { return termText; }
+ public final char[] termBuffer() { return termBuffer; }
+ public final int termBufferOffset() { return termBufferOffset; }
+ public final int termBufferLength() { return termBufferLength; }
+ public void setStartOffset(int offset) {this.startOffset = offset;}
+ public void setEndOffset(int offset) {this.endOffset = offset;}
+
+ public final void setTermBuffer(char[] buffer, int offset, int length) {
+ this.termBuffer = buffer;
+ this.termBufferOffset = offset;
+ this.termBufferLength = length;
+ }
+
+
/** Returns this Token's starting offset, the position of the first character
corresponding to this token in the source text.
Index: src/java/org/apache/lucene/index/FieldInfos.java
===================================================================
--- src/java/org/apache/lucene/index/FieldInfos.java (revision 533617)
+++ src/java/org/apache/lucene/index/FieldInfos.java (working copy)
@@ -156,7 +156,7 @@
* @param omitNorms true if the norms for the indexed field should be omitted
*/
public void add(String name, boolean isIndexed, boolean storeTermVector,
- boolean storePositionWithTermVector, boolean storeOffsetWithTermVector, boolean omitNorms) {
+ boolean storePositionWithTermVector, boolean storeOffsetWithTermVector, boolean omitNorms) {
add(name, isIndexed, storeTermVector, storePositionWithTermVector,
storeOffsetWithTermVector, omitNorms, false);
}
@@ -174,12 +174,13 @@
* @param omitNorms true if the norms for the indexed field should be omitted
* @param storePayloads true if payloads should be stored for this field
*/
- public void add(String name, boolean isIndexed, boolean storeTermVector,
- boolean storePositionWithTermVector, boolean storeOffsetWithTermVector,
- boolean omitNorms, boolean storePayloads) {
+ // nocommit: API change, but, not yet released
+ public FieldInfo add(String name, boolean isIndexed, boolean storeTermVector,
+ boolean storePositionWithTermVector, boolean storeOffsetWithTermVector,
+ boolean omitNorms, boolean storePayloads) {
FieldInfo fi = fieldInfo(name);
if (fi == null) {
- addInternal(name, isIndexed, storeTermVector, storePositionWithTermVector, storeOffsetWithTermVector, omitNorms, storePayloads);
+ return addInternal(name, isIndexed, storeTermVector, storePositionWithTermVector, storeOffsetWithTermVector, omitNorms, storePayloads);
} else {
if (fi.isIndexed != isIndexed) {
fi.isIndexed = true; // once indexed, always index
@@ -199,19 +200,20 @@
if (fi.storePayloads != storePayloads) {
fi.storePayloads = true;
}
-
}
+ return fi;
}
-
- private void addInternal(String name, boolean isIndexed,
- boolean storeTermVector, boolean storePositionWithTermVector,
- boolean storeOffsetWithTermVector, boolean omitNorms, boolean storePayloads) {
+ // nocommit: API change
+ private FieldInfo addInternal(String name, boolean isIndexed,
+ boolean storeTermVector, boolean storePositionWithTermVector,
+ boolean storeOffsetWithTermVector, boolean omitNorms, boolean storePayloads) {
FieldInfo fi =
new FieldInfo(name, isIndexed, byNumber.size(), storeTermVector, storePositionWithTermVector,
storeOffsetWithTermVector, omitNorms, storePayloads);
byNumber.add(fi);
byName.put(name, fi);
+ return fi;
}
public int fieldNumber(String fieldName) {
Index: src/java/org/apache/lucene/index/IndexReader.java
===================================================================
--- src/java/org/apache/lucene/index/IndexReader.java (revision 533617)
+++ src/java/org/apache/lucene/index/IndexReader.java (working copy)
@@ -771,7 +771,7 @@
// KeepOnlyLastCommitDeleter:
IndexFileDeleter deleter = new IndexFileDeleter(directory,
deletionPolicy == null ? new KeepOnlyLastCommitDeletionPolicy() : deletionPolicy,
- segmentInfos, null);
+ segmentInfos, null, null);
// Checkpoint the state we are about to change, in
// case we have to roll back:
Index: src/java/org/apache/lucene/index/IndexFileNames.java
===================================================================
--- src/java/org/apache/lucene/index/IndexFileNames.java (revision 533617)
+++ src/java/org/apache/lucene/index/IndexFileNames.java (working copy)
@@ -60,7 +60,7 @@
*/
static final String INDEX_EXTENSIONS[] = new String[] {
"cfs", "fnm", "fdx", "fdt", "tii", "tis", "frq", "prx", "del",
- "tvx", "tvd", "tvf", "gen", "nrm"
+ "tvx", "tvd", "tvf", "gen", "nrm",
};
/** File extensions that are added to a compound file
Index: src/java/org/apache/lucene/index/TermVectorsReader.java
===================================================================
--- src/java/org/apache/lucene/index/TermVectorsReader.java (revision 533617)
+++ src/java/org/apache/lucene/index/TermVectorsReader.java (working copy)
@@ -90,6 +90,7 @@
TermFreqVector get(int docNum, String field) throws IOException {
// Check if no term vectors are available for this segment at all
int fieldNumber = fieldInfos.fieldNumber(field);
+ //System.out.println(" tvr.get: docNum=" + docNum);
TermFreqVector result = null;
if (tvx != null) {
//We need to account for the FORMAT_SIZE at when seeking in the tvx
@@ -145,6 +146,7 @@
*/
TermFreqVector[] get(int docNum) throws IOException {
TermFreqVector[] result = null;
+ //System.out.println("get TV docNum=" + docNum);
// Check if no term vectors are available for this segment at all
if (tvx != null) {
//We need to offset by
@@ -208,6 +210,8 @@
//We don't need to offset by the FORMAT here since the pointer already includes the offset
tvf.seek(tvfPointer);
+ //System.out.println("read TV field=" + field + " tvfPointer=" + tvfPointer);
+
int numTerms = tvf.readVInt();
//System.out.println("Num Terms: " + numTerms);
// If no terms - return a constant empty termvector. However, this should never occur!
@@ -244,11 +248,14 @@
int totalLength = 0;
char [] buffer = new char[10]; // init the buffer with a length of 10 character
char[] previousBuffer = {};
+
+ //System.out.println(" numTerms=" + numTerms + " storePos=" + storePositions + " storeOffs=" + storeOffsets);
for (int i = 0; i < numTerms; i++) {
start = tvf.readVInt();
deltaLength = tvf.readVInt();
totalLength = start + deltaLength;
+ //System.out.println(" prefix=" + start + " suffix=" + deltaLength);
if (buffer.length < totalLength) { // increase buffer
buffer = null; // give a hint to garbage collector
buffer = new char[totalLength];
@@ -259,9 +266,11 @@
tvf.readChars(buffer, start, deltaLength);
terms[i] = new String(buffer, 0, totalLength);
+ //System.out.println(" term=" + terms[i]);
previousBuffer = buffer;
int freq = tvf.readVInt();
termFreqs[i] = freq;
+ //System.out.println(" freq=" + freq);
if (storePositions) { //read in the positions
int [] pos = new int[freq];
@@ -270,6 +279,7 @@
for (int j = 0; j < freq; j++)
{
pos[j] = prevPosition + tvf.readVInt();
+ //System.out.println(" read pos=" + pos[j]);
prevPosition = pos[j];
}
}
@@ -281,6 +291,7 @@
for (int j = 0; j < freq; j++) {
int startOffset = prevOffset + tvf.readVInt();
int endOffset = startOffset + tvf.readVInt();
+ //System.out.println(" read offs=" + startOffset + " to " + endOffset);
offs[j] = new TermVectorOffsetInfo(startOffset, endOffset);
prevOffset = endOffset;
}
Index: src/java/org/apache/lucene/index/SegmentTermDocs.java
===================================================================
--- src/java/org/apache/lucene/index/SegmentTermDocs.java (revision 533617)
+++ src/java/org/apache/lucene/index/SegmentTermDocs.java (working copy)
@@ -51,6 +51,7 @@
}
public void seek(Term term) throws IOException {
+ //System.out.println("std: seek term " + term);
TermInfo ti = parent.tis.get(term);
seek(ti, term);
}
@@ -74,6 +75,7 @@
void seek(TermInfo ti, Term term) throws IOException {
count = 0;
+ //System.out.println("std: seek " + ti);
payloadLengthAtLastSkip = 0;
FieldInfo fi = parent.fieldInfos.fieldInfo(term.field);
currentFieldStoresPayloads = (fi != null) ? fi.storePayloads : false;
@@ -81,6 +83,7 @@
df = 0;
} else {
df = ti.docFreq;
+ //System.out.println(" df = " + df);
doc = 0;
skipDoc = 0;
skipCount = 0;
@@ -90,6 +93,7 @@
skipPointer = freqPointer + ti.skipOffset;
freqStream.seek(freqPointer);
haveSkipped = false;
+ //System.out.println(" freqP = " + freqPointer + "; proxP = " + proxPointer);
}
}
@@ -106,6 +110,7 @@
}
public boolean next() throws IOException {
+ //System.out.println("std: now do next");
while (true) {
if (count == df)
return false;
@@ -118,6 +123,7 @@
freq = freqStream.readVInt(); // else read freq
count++;
+ //System.out.println("std: read " + docCode + "; freq= " + freq);
if (deletedDocs == null || !deletedDocs.get(doc))
break;
@@ -131,18 +137,22 @@
throws IOException {
final int length = docs.length;
int i = 0;
+ //System.out.println("read bulk: df =" + df + "; length=" + length);
while (i < length && count < df) {
// manually inlined call to next() for speed
final int docCode = freqStream.readVInt();
+ //System.out.println(" read code: " + docCode);
doc += docCode >>> 1; // shift off low bit
if ((docCode & 1) != 0) // if low bit is set
freq = 1; // freq is one
else
freq = freqStream.readVInt(); // else read freq
count++;
+ // //System.out.println(" read freq " + freq);
if (deletedDocs == null || !deletedDocs.get(doc)) {
+ //System.out.println(" add " + doc + "; freq=" + freq);
docs[i] = doc;
freqs[i] = freq;
++i;
@@ -156,7 +166,9 @@
/** Optimized implementation. */
public boolean skipTo(int target) throws IOException {
+ //System.out.println("std skip to " + target);
if (df >= skipInterval) { // optimized case
+ //System.out.println(" is frequent enough");
if (skipStream == null)
skipStream = (IndexInput) freqStream.clone(); // lazily clone
@@ -172,6 +184,7 @@
long lastFreqPointer = freqStream.getFilePointer();
long lastProxPointer = -1;
int numSkipped = -1 - (count % skipInterval);
+ //System.out.println(" target " + target + "; skipDoc " + skipDoc);
while (target > skipDoc) {
lastSkipDoc = skipDoc;
@@ -203,11 +216,13 @@
freqPointer += skipStream.readVInt();
proxPointer += skipStream.readVInt();
+ //System.out.println(" now freq " + freqPointer + " prox " + proxPointer);
skipCount++;
}
// if we found something to skip, then skip it
if (lastFreqPointer > freqStream.getFilePointer()) {
+ //System.out.println(" do skip! " + lastFreqPointer);
freqStream.seek(lastFreqPointer);
skipProx(lastProxPointer, lastPayloadLength);
@@ -219,6 +234,7 @@
// done skipping, now just scan
do {
+ //System.out.println(" now scan " + target + " " + doc);
if (!next())
return false;
} while (target > doc);
Index: src/java/org/apache/lucene/index/FieldsWriter.java
===================================================================
--- src/java/org/apache/lucene/index/FieldsWriter.java (revision 533617)
+++ src/java/org/apache/lucene/index/FieldsWriter.java (working copy)
@@ -38,17 +38,90 @@
private IndexOutput indexStream;
+ private boolean doClose;
+
FieldsWriter(Directory d, String segment, FieldInfos fn) throws IOException {
fieldInfos = fn;
fieldsStream = d.createOutput(segment + ".fdt");
indexStream = d.createOutput(segment + ".fdx");
+ doClose = true;
}
+ FieldsWriter(IndexOutput fdx, IndexOutput fdt, FieldInfos fn) throws IOException {
+ fieldInfos = fn;
+ fieldsStream = fdt;
+ indexStream = fdx;
+ doClose = false;
+ }
+ IndexOutput getIndexStream() {
+ return indexStream;
+ }
+ IndexOutput getFieldsStream() {
+ return fieldsStream;
+ }
+
final void close() throws IOException {
+ if (doClose) {
fieldsStream.close();
indexStream.close();
+ }
}
+ final void writeField(FieldInfo fi, Fieldable field) throws IOException {
+ // if the field as an instanceof FieldsReader.FieldForMerge, we're in merge mode
+ // and field.binaryValue() already returns the compressed value for a field
+ // with isCompressed()==true, so we disable compression in that case
+ boolean disableCompression = (field instanceof FieldsReader.FieldForMerge);
+ fieldsStream.writeVInt(fi.number);
+ // System.out.println(" write field number " + fieldInfos.fieldNumber(field.name()) + " name " + field.name() + " to " + fieldsStream + " at " + fieldsStream.getFilePointer());
+ byte bits = 0;
+ if (field.isTokenized())
+ bits |= FieldsWriter.FIELD_IS_TOKENIZED;
+ if (field.isBinary())
+ bits |= FieldsWriter.FIELD_IS_BINARY;
+ if (field.isCompressed())
+ bits |= FieldsWriter.FIELD_IS_COMPRESSED;
+
+ fieldsStream.writeByte(bits);
+
+ if (field.isCompressed()) {
+ // compression is enabled for the current field
+ byte[] data = null;
+
+ if (disableCompression) {
+ // optimized case for merging, the data
+ // is already compressed
+ data = field.binaryValue();
+ } else {
+ // check if it is a binary field
+ if (field.isBinary()) {
+ data = compress(field.binaryValue());
+ }
+ else {
+ data = compress(field.stringValue().getBytes("UTF-8"));
+ }
+ }
+ final int len = data.length;
+ // System.out.println(" compressed: " + len);
+ fieldsStream.writeVInt(len);
+ fieldsStream.writeBytes(data, len);
+ }
+ else {
+ // compression is disabled for the current field
+ if (field.isBinary()) {
+ byte[] data = field.binaryValue();
+ final int len = data.length;
+ // System.out.println(" not compressed: " + len);
+ fieldsStream.writeVInt(len);
+ fieldsStream.writeBytes(data, len);
+ }
+ else {
+ fieldsStream.writeString(field.stringValue());
+ }
+ }
+ // System.out.println(" fieldsStream now at " + fieldsStream.getFilePointer());
+ }
+
final void addDocument(Document doc) throws IOException {
indexStream.writeLong(fieldsStream.getFilePointer());
@@ -59,62 +132,14 @@
if (field.isStored())
storedCount++;
}
+ // System.out.println("write " + storedCount + " fields to " + fieldsStream + " at " + fieldsStream.getFilePointer());
fieldsStream.writeVInt(storedCount);
fieldIterator = doc.getFields().iterator();
while (fieldIterator.hasNext()) {
Fieldable field = (Fieldable) fieldIterator.next();
- // if the field as an instanceof FieldsReader.FieldForMerge, we're in merge mode
- // and field.binaryValue() already returns the compressed value for a field
- // with isCompressed()==true, so we disable compression in that case
- boolean disableCompression = (field instanceof FieldsReader.FieldForMerge);
- if (field.isStored()) {
- fieldsStream.writeVInt(fieldInfos.fieldNumber(field.name()));
-
- byte bits = 0;
- if (field.isTokenized())
- bits |= FieldsWriter.FIELD_IS_TOKENIZED;
- if (field.isBinary())
- bits |= FieldsWriter.FIELD_IS_BINARY;
- if (field.isCompressed())
- bits |= FieldsWriter.FIELD_IS_COMPRESSED;
-
- fieldsStream.writeByte(bits);
-
- if (field.isCompressed()) {
- // compression is enabled for the current field
- byte[] data = null;
-
- if (disableCompression) {
- // optimized case for merging, the data
- // is already compressed
- data = field.binaryValue();
- } else {
- // check if it is a binary field
- if (field.isBinary()) {
- data = compress(field.binaryValue());
- }
- else {
- data = compress(field.stringValue().getBytes("UTF-8"));
- }
- }
- final int len = data.length;
- fieldsStream.writeVInt(len);
- fieldsStream.writeBytes(data, len);
- }
- else {
- // compression is disabled for the current field
- if (field.isBinary()) {
- byte[] data = field.binaryValue();
- final int len = data.length;
- fieldsStream.writeVInt(len);
- fieldsStream.writeBytes(data, len);
- }
- else {
- fieldsStream.writeString(field.stringValue());
- }
- }
- }
+ if (field.isStored())
+ writeField(fieldInfos.fieldInfo(field.name()), field);
}
}
Index: src/java/org/apache/lucene/index/SegmentMerger.java
===================================================================
--- src/java/org/apache/lucene/index/SegmentMerger.java (revision 533617)
+++ src/java/org/apache/lucene/index/SegmentMerger.java (working copy)
@@ -289,6 +289,7 @@
top = (SegmentMergeInfo) queue.top();
}
+ //System.out.println("merge term=" + term);
mergeTermInfo(match, matchSize); // add new TermInfo
while (matchSize > 0) {
@@ -348,6 +349,7 @@
boolean storePayloads = fieldInfos.fieldInfo(smis[0].term.field).storePayloads;
int lastPayloadLength = -1; // ensures that we write the first length
for (int i = 0; i < n; i++) {
+ //System.out.println(" merge seg=" + i + " of " + n);
SegmentMergeInfo smi = smis[i];
TermPositions postings = smi.getPositions();
int base = smi.base;
@@ -355,6 +357,7 @@
postings.seek(smi.termEnum);
while (postings.next()) {
int doc = postings.doc();
+ //System.out.println(" doc=" + doc);
if (docMap != null)
doc = docMap[doc]; // map around deletions
doc += base; // convert to merged space
Index: src/java/org/apache/lucene/index/IndexWriter.java
===================================================================
--- src/java/org/apache/lucene/index/IndexWriter.java (revision 533617)
+++ src/java/org/apache/lucene/index/IndexWriter.java (working copy)
@@ -60,10 +60,11 @@
(which just deletes and then adds). When finished adding, deleting and updating documents, close should be called.
These changes are buffered in memory and periodically
- flushed to the {@link Directory} (during the above method calls). A flush is triggered when there are
- enough buffered deletes (see {@link
- #setMaxBufferedDeleteTerms}) or enough added documents
- (see {@link #setMaxBufferedDocs}) since the last flush,
+ flushed to the {@link Directory} (during the above method
+ calls). A flush is triggered when there are enough
+ buffered deletes (see {@link #setMaxBufferedDeleteTerms})
+ or enough added documents (see {@link #setMaxBufferedDocs}
+ and {@link #setRAMBufferSizeMB}) since the last flush,
whichever is sooner. When a flush occurs, both pending
deletes and added documents are flushed to the index. A
flush may also trigger one or more segment merges.
@@ -171,11 +172,17 @@
public final static int DEFAULT_MERGE_FACTOR = 10;
/**
- * Default value is 10. Change using {@link #setMaxBufferedDocs(int)}.
+ * Default value is 0 (meaning flush is based on RAM usage
+ * by default). Change using {@link #setMaxBufferedDocs}.
*/
- public final static int DEFAULT_MAX_BUFFERED_DOCS = 10;
+ public final static int DEFAULT_MAX_BUFFERED_DOCS = 0;
/**
+ * Default value is 16 MB. Change using {@link #setRAMBufferSizeMB}.
+ */
+ public final static float DEFAULT_RAM_BUFFER_SIZE_MB = 16F;
+
+ /**
* Default value is 1000. Change using {@link #setMaxBufferedDeleteTerms(int)}.
*/
public final static int DEFAULT_MAX_BUFFERED_DELETE_TERMS = 1000;
@@ -208,8 +215,7 @@
private boolean autoCommit = true; // false if we should commit only on close
SegmentInfos segmentInfos = new SegmentInfos(); // the segments
- SegmentInfos ramSegmentInfos = new SegmentInfos(); // the segments in ramDirectory
- private final RAMDirectory ramDirectory = new RAMDirectory(); // for temp segs
+ private MultiDocumentWriter docWriter;
private IndexFileDeleter deleter;
private Lock writeLock;
@@ -563,6 +569,10 @@
}
}
+ IndexFileDeleter getDeleter() {
+ return deleter;
+ }
+
private void init(Directory d, Analyzer a, final boolean create, boolean closeDir, IndexDeletionPolicy deletionPolicy, boolean autoCommit)
throws CorruptIndexException, LockObtainFailedException, IOException {
this.closeDir = closeDir;
@@ -602,11 +612,14 @@
rollbackSegmentInfos = (SegmentInfos) segmentInfos.clone();
}
+ docWriter = new MultiDocumentWriter(newSegmentName(), directory, this, !autoCommit);
+ docWriter.setInfoStream(infoStream);
+
// Default deleter (for backwards compatibility) is
// KeepOnlyLastCommitDeleter:
deleter = new IndexFileDeleter(directory,
deletionPolicy == null ? new KeepOnlyLastCommitDeletionPolicy() : deletionPolicy,
- segmentInfos, infoStream);
+ segmentInfos, infoStream, docWriter);
} catch (IOException e) {
this.writeLock.release();
@@ -661,19 +674,26 @@
}
/** Determines the minimal number of documents required before the buffered
- * in-memory documents are merged and a new Segment is created.
+ * in-memory documents are flushed as a new Segment.
* Since Documents are merged in a {@link org.apache.lucene.store.RAMDirectory},
* large value gives faster indexing. At the same time, mergeFactor limits
* the number of files open in a FSDirectory.
*
- * The default value is 10.
+ *
If this is 0, then the RAM buffer is flushed instead
+ * by overally RAM usage (see {@link
+ * #setRAMBufferSizeMB}). If this is non-zero, then
+ * flushing is triggered by maxBufferedDocs and not by
+ * overall RAM usage.
*
- * @throws IllegalArgumentException if maxBufferedDocs is smaller than 2
+ * The default value is 0.
+ *
+ * @throws IllegalArgumentException if maxBufferedDocs is
+ * non-zero and smaller than 2
*/
public void setMaxBufferedDocs(int maxBufferedDocs) {
ensureOpen();
- if (maxBufferedDocs < 2)
- throw new IllegalArgumentException("maxBufferedDocs must at least be 2");
+ if (maxBufferedDocs != 0 && maxBufferedDocs < 2)
+ throw new IllegalArgumentException("maxBufferedDocs must at least be 2 or 0 to disable");
this.minMergeDocs = maxBufferedDocs;
}
@@ -685,7 +705,28 @@
return minMergeDocs;
}
+ /** Determines the amount of RAM that may be used for
+ * buffering before the in-memory documents are flushed as
+ * a new Segment. This only applies when maxBufferedDocs
+ * is set to 0. Generally for faster indexing performance
+ * it's best to flush by RAM usage instead of document
+ * count.
+ */
+ public void setRAMBufferSizeMB(float mb) {
+ if (mb < 1)
+ throw new IllegalArgumentException("ramBufferSize must at least be 1 MB");
+ ramBufferSize = mb*1024F*1024F;
+ docWriter.setRAMBufferSizeMB(mb);
+ }
+
/**
+ * @see #setRAMBufferSizeMB
+ */
+ public float getRAMBufferSizeMB() {
+ return ramBufferSize/1024F/1024F;
+ }
+
+ /**
* Determines the minimal number of delete terms required before the buffered
* in-memory delete terms are applied and flushed. If there are documents
* buffered in memory at the time, they are merged and a new segment is
@@ -756,7 +797,9 @@
public void setInfoStream(PrintStream infoStream) {
ensureOpen();
this.infoStream = infoStream;
- deleter.setInfoStream(infoStream);
+ docWriter.setInfoStream(infoStream);
+ // nocommit
+ //deleter.setInfoStream(infoStream);
}
/**
@@ -835,7 +878,7 @@
*/
public synchronized void close() throws CorruptIndexException, IOException {
if (!closed) {
- flushRamSegments();
+ flush();
if (commitPending) {
segmentInfos.write(directory); // now commit changes
@@ -844,7 +887,6 @@
rollbackSegmentInfos = null;
}
- ramDirectory.close();
if (writeLock != null) {
writeLock.release(); // release write lock
writeLock = null;
@@ -884,7 +926,7 @@
/** Returns the number of documents currently in this index. */
public synchronized int docCount() {
ensureOpen();
- int count = ramSegmentInfos.size();
+ int count = docWriter.docID;
for (int i = 0; i < segmentInfos.size(); i++) {
SegmentInfo si = segmentInfos.info(i);
count += si.docCount;
@@ -962,24 +1004,15 @@
*/
public void addDocument(Document doc, Analyzer analyzer) throws CorruptIndexException, IOException {
ensureOpen();
- SegmentInfo newSegmentInfo = buildSingleDocSegment(doc, analyzer);
- synchronized (this) {
- ramSegmentInfos.addElement(newSegmentInfo);
- maybeFlushRamSegments();
+ synchronized(this) {
+ docWriter.addDocument(doc, analyzer);
+ // For the non-autoCommit case, MultiDocumentWriter
+ // takes care of flushing its pending state to disk
+ if (autoCommit)
+ maybeFlush();
}
}
- SegmentInfo buildSingleDocSegment(Document doc, Analyzer analyzer)
- throws CorruptIndexException, IOException {
- DocumentWriter dw = new DocumentWriter(ramDirectory, analyzer, this);
- dw.setInfoStream(infoStream);
- String segmentName = newRamSegmentName();
- dw.addDocument(segmentName, doc);
- SegmentInfo si = new SegmentInfo(segmentName, 1, ramDirectory, false, false);
- si.setNumFields(dw.getNumFields());
- return si;
- }
-
/**
* Deletes the document(s) containing term.
* @param term the term to identify the documents to be deleted
@@ -989,7 +1022,7 @@
public synchronized void deleteDocuments(Term term) throws CorruptIndexException, IOException {
ensureOpen();
bufferDeleteTerm(term);
- maybeFlushRamSegments();
+ maybeFlush();
}
/**
@@ -1005,7 +1038,7 @@
for (int i = 0; i < terms.length; i++) {
bufferDeleteTerm(terms[i]);
}
- maybeFlushRamSegments();
+ maybeFlush();
}
/**
@@ -1041,26 +1074,25 @@
public void updateDocument(Term term, Document doc, Analyzer analyzer)
throws CorruptIndexException, IOException {
ensureOpen();
- SegmentInfo newSegmentInfo = buildSingleDocSegment(doc, analyzer);
- synchronized (this) {
- bufferDeleteTerm(term);
- ramSegmentInfos.addElement(newSegmentInfo);
- maybeFlushRamSegments();
+ bufferDeleteTerm(term);
+ synchronized(this) {
+ docWriter.addDocument(doc, analyzer);
+ // nocommit: what if we need to trigger on max delete terms?
+ // For the non-autoCommit case, MultiDocumentWriter
+ // takes care of flushing its pending state to disk
+ if (autoCommit)
+ maybeFlush();
}
}
- final synchronized String newRamSegmentName() {
- return "_ram_" + Integer.toString(ramSegmentInfos.counter++, Character.MAX_RADIX);
- }
-
// for test purpose
final synchronized int getSegmentCount(){
return segmentInfos.size();
}
// for test purpose
- final synchronized int getRamSegmentCount(){
- return ramSegmentInfos.size();
+ final synchronized int getNumBufferedDocuments(){
+ return docWriter.docID;
}
// for test purpose
@@ -1089,6 +1121,7 @@
*/
private int mergeFactor = DEFAULT_MERGE_FACTOR;
+ // nocommit fix javadocs
/** Determines the minimal number of documents required before the buffered
* in-memory documents are merging and a new Segment is created.
* Since Documents are merged in a {@link org.apache.lucene.store.RAMDirectory},
@@ -1096,10 +1129,11 @@
* the number of files open in a FSDirectory.
*
*
The default value is {@link #DEFAULT_MAX_BUFFERED_DOCS}.
-
*/
private int minMergeDocs = DEFAULT_MAX_BUFFERED_DOCS;
+ // nocommit javadoc
+ private float ramBufferSize = DEFAULT_RAM_BUFFER_SIZE_MB*1024F*1024F;
/** Determines the largest number of documents ever merged by addDocument().
* Small values (e.g., less than 10,000) are best for interactive indexing,
@@ -1183,7 +1217,7 @@
*/
public synchronized void optimize() throws CorruptIndexException, IOException {
ensureOpen();
- flushRamSegments();
+ flush();
while (segmentInfos.size() > 1 ||
(segmentInfos.size() == 1 &&
(SegmentReader.hasDeletions(segmentInfos.info(0)) ||
@@ -1192,7 +1226,7 @@
(useCompoundFile &&
(!SegmentReader.usesCompoundFile(segmentInfos.info(0))))))) {
int minSegment = segmentInfos.size() - mergeFactor;
- mergeSegments(segmentInfos, minSegment < 0 ? 0 : minSegment, segmentInfos.size());
+ mergeSegments(minSegment < 0 ? 0 : minSegment, segmentInfos.size());
}
}
@@ -1209,7 +1243,7 @@
localRollbackSegmentInfos = (SegmentInfos) segmentInfos.clone();
localAutoCommit = autoCommit;
if (localAutoCommit) {
- flushRamSegments();
+ flush();
// Turn off auto-commit during our local transaction:
autoCommit = false;
} else
@@ -1299,16 +1333,18 @@
segmentInfos.clear();
segmentInfos.addAll(rollbackSegmentInfos);
+ docWriter.abort();
+
// Ask deleter to locate unreferenced files & remove
// them:
deleter.checkpoint(segmentInfos, false);
deleter.refresh();
- ramSegmentInfos = new SegmentInfos();
bufferedDeleteTerms.clear();
numBufferedDeleteTerms = 0;
commitPending = false;
+ docWriter.abort();
close();
} else {
@@ -1403,7 +1439,7 @@
for (int base = start; base < segmentInfos.size(); base++) {
int end = Math.min(segmentInfos.size(), base+mergeFactor);
if (end-base > 1) {
- mergeSegments(segmentInfos, base, end);
+ mergeSegments(base, end);
}
}
}
@@ -1443,7 +1479,7 @@
// segments in S may not since they could come from multiple indexes.
// Here is the merge algorithm for addIndexesNoOptimize():
//
- // 1 Flush ram segments.
+ // 1 Flush ram.
// 2 Consider a combined sequence with segments from T followed
// by segments from S (same as current addIndexes(Directory[])).
// 3 Assume the highest level for segments in S is h. Call
@@ -1464,14 +1500,18 @@
// copy a segment, which may cause doc count to change because deleted
// docs are garbage collected.
- // 1 flush ram segments
+ // 1 flush ram
ensureOpen();
- flushRamSegments();
+ flush();
// 2 copy segment infos and find the highest level from dirs
int startUpperBound = minMergeDocs;
+ // nocommit: what to do?
+ if (startUpperBound == 0)
+ startUpperBound = 10;
+
boolean success = false;
startTransaction();
@@ -1530,7 +1570,7 @@
// copy those segments from S
for (int i = segmentCount - numSegmentsToCopy; i < segmentCount; i++) {
- mergeSegments(segmentInfos, i, i + 1);
+ mergeSegments(i, i + 1);
}
if (checkNonDecreasingLevels(segmentCount - numSegmentsToCopy)) {
success = true;
@@ -1539,7 +1579,7 @@
}
// invariants do not hold, simply merge those segments
- mergeSegments(segmentInfos, segmentCount - numTailSegments, segmentCount);
+ mergeSegments(segmentCount - numTailSegments, segmentCount);
// maybe merge segments again if necessary
if (segmentInfos.info(segmentInfos.size() - 1).docCount > startUpperBound) {
@@ -1679,19 +1719,23 @@
throws IOException {
}
- protected final void maybeFlushRamSegments() throws CorruptIndexException, IOException {
+ protected final synchronized void maybeFlush() throws CorruptIndexException, IOException {
// A flush is triggered if enough new documents are buffered or
- // if enough delete terms are buffered
- if (ramSegmentInfos.size() >= minMergeDocs || numBufferedDeleteTerms >= maxBufferedDeleteTerms) {
- flushRamSegments();
- }
- }
+ // if enough delete terms are buffered or enough RAM is
+ // being consumed
+ // nocommit
+ if (numBufferedDeleteTerms >= maxBufferedDeleteTerms ||
+ (autoCommit && ((minMergeDocs != 0 && docWriter.docID >= minMergeDocs) ||
+ ((true || minMergeDocs == 0) && autoCommit && docWriter.getRAMUsed() > ramBufferSize)))) {
- /** Expert: Flushes all RAM-resident segments (buffered documents), then may merge segments. */
- private final synchronized void flushRamSegments() throws CorruptIndexException, IOException {
- if (ramSegmentInfos.size() > 0 || bufferedDeleteTerms.size() > 0) {
- mergeSegments(ramSegmentInfos, 0, ramSegmentInfos.size());
- maybeMergeSegments(minMergeDocs);
+ /*
+ if (minMergeDocs < docWriter.docID)
+ throw new RuntimeException("too small minMergeDocs=" + minMergeDocs + " #docs=" + docWriter.docID);
+ if (minMergeDocs > 6*docWriter.docID)
+ throw new RuntimeException("too large minMergeDocs=" + minMergeDocs + " #docs=" + docWriter.docID);
+ */
+
+ flush();
}
}
@@ -1705,7 +1749,86 @@
*/
public final synchronized void flush() throws CorruptIndexException, IOException {
ensureOpen();
- flushRamSegments();
+
+ SegmentInfo newSegment = null;
+ boolean anything = false;
+
+ boolean flushDocs = docWriter.docID > 0;
+ boolean flushDeletes = bufferedDeleteTerms.size() > 0;
+ final int numDocs = docWriter.docID;
+
+ if (flushDocs || flushDeletes) {
+
+ SegmentInfos rollback = null;
+
+ if (flushDeletes)
+ rollback = (SegmentInfos) segmentInfos.clone();
+
+ boolean success = false;
+
+ try {
+ if (flushDocs) {
+ int mergedDocCount = docWriter.docID;
+ String segment = docWriter.segment;
+ docWriter.flush(newSegmentName());
+ newSegment = new SegmentInfo(segment,
+ mergedDocCount,
+ directory, false, true);
+ segmentInfos.addElement(newSegment);
+ }
+
+ if (flushDeletes) {
+ maybeApplyDeletes(flushDocs);
+ doAfterFlush();
+ }
+
+ checkpoint();
+ success = true;
+ } finally {
+ if (!success) {
+ if (flushDeletes) {
+ // Fully replace the segmentInfos since flushed
+ // deletes could have changed any of the
+ // SegmentInfo instances:
+ segmentInfos.clear();
+ segmentInfos.addAll(rollback);
+ } else {
+ // Remove segment we added, if any:
+ if (newSegment != null &&
+ segmentInfos.size() > 0 &&
+ segmentInfos.info(segmentInfos.size()-1) == newSegment)
+ segmentInfos.remove(segmentInfos.size()-1);
+ docWriter.abort();
+ }
+ deleter.checkpoint(segmentInfos, false);
+ deleter.refresh();
+ }
+ }
+
+ deleter.checkpoint(segmentInfos, autoCommit);
+
+ if (flushDocs && useCompoundFile) {
+ success = false;
+ try {
+ docWriter.createCompoundFile(newSegment.name);
+ newSegment.setUseCompoundFile(true);
+ checkpoint();
+ success = true;
+ } finally {
+ if (!success) {
+ newSegment.setUseCompoundFile(false);
+ deleter.refresh();
+ }
+ }
+
+ deleter.checkpoint(segmentInfos, autoCommit);
+ }
+
+ // nocommit
+ // maybeMergeSegments(mergeFactor * numDocs / 2);
+
+ maybeMergeSegments(minMergeDocs);
+ }
}
/** Expert: Return the total size of all index files currently cached in memory.
@@ -1713,15 +1836,15 @@
*/
public final long ramSizeInBytes() {
ensureOpen();
- return ramDirectory.sizeInBytes();
+ return docWriter.getRAMUsed();
}
/** Expert: Return the number of documents whose segments are currently cached in memory.
- * Useful when calling flushRamSegments()
+ * Useful when calling flush()
*/
public final synchronized int numRamDocs() {
ensureOpen();
- return ramSegmentInfos.size();
+ return docWriter.docID;
}
/** Incremental segment merger. */
@@ -1729,6 +1852,9 @@
long lowerBound = -1;
long upperBound = startUpperBound;
+ // nocommit
+ if (upperBound == 0) upperBound = 10;
+
while (upperBound < maxMergeDocs) {
int minSegment = segmentInfos.size();
int maxSegment = -1;
@@ -1760,7 +1886,7 @@
while (numSegments >= mergeFactor) {
// merge the leftmost* mergeFactor segments
- int docCount = mergeSegments(segmentInfos, minSegment, minSegment + mergeFactor);
+ int docCount = mergeSegments(minSegment, minSegment + mergeFactor);
numSegments -= mergeFactor;
if (docCount > upperBound) {
@@ -1789,39 +1915,33 @@
* Merges the named range of segments, replacing them in the stack with a
* single segment.
*/
- private final int mergeSegments(SegmentInfos sourceSegments, int minSegment, int end)
+ long netMergeTime;
+
+ private final int mergeSegments(int minSegment, int end)
throws CorruptIndexException, IOException {
- // We may be called solely because there are deletes
- // pending, in which case doMerge is false:
- boolean doMerge = end > 0;
final String mergedName = newSegmentName();
+
+ long t0 = System.currentTimeMillis();
+
SegmentMerger merger = null;
-
- final List ramSegmentsToDelete = new ArrayList();
-
SegmentInfo newSegment = null;
int mergedDocCount = 0;
- boolean anyDeletes = (bufferedDeleteTerms.size() != 0);
// This is try/finally to make sure merger's readers are closed:
try {
- if (doMerge) {
- if (infoStream != null) infoStream.print("merging segments");
- merger = new SegmentMerger(this, mergedName);
+ if (infoStream != null) infoStream.print("merging segments");
- for (int i = minSegment; i < end; i++) {
- SegmentInfo si = sourceSegments.info(i);
- if (infoStream != null)
- infoStream.print(" " + si.name + " (" + si.docCount + " docs)");
- IndexReader reader = SegmentReader.get(si); // no need to set deleter (yet)
- merger.add(reader);
- if (reader.directory() == this.ramDirectory) {
- ramSegmentsToDelete.add(si);
- }
- }
+ merger = new SegmentMerger(this, mergedName);
+
+ for (int i = minSegment; i < end; i++) {
+ SegmentInfo si = segmentInfos.info(i);
+ if (infoStream != null)
+ infoStream.print(" " + si.name + " (" + si.docCount + " docs)");
+ IndexReader reader = SegmentReader.get(si); // no need to set deleter (yet)
+ merger.add(reader);
}
SegmentInfos rollback = null;
@@ -1831,99 +1951,57 @@
// if we hit exception when doing the merge:
try {
- if (doMerge) {
- mergedDocCount = merger.merge();
+ mergedDocCount = merger.merge();
- if (infoStream != null) {
- infoStream.println(" into "+mergedName+" ("+mergedDocCount+" docs)");
- }
+ if (infoStream != null) {
+ infoStream.println(" into "+mergedName+" ("+mergedDocCount+" docs)");
+ }
- newSegment = new SegmentInfo(mergedName, mergedDocCount,
- directory, false, true);
- }
+ newSegment = new SegmentInfo(mergedName, mergedDocCount,
+ directory, false, true);
- if (sourceSegments != ramSegmentInfos || anyDeletes) {
- // Now save the SegmentInfo instances that
- // we are replacing:
- rollback = (SegmentInfos) segmentInfos.clone();
- }
+ rollback = (SegmentInfos) segmentInfos.clone();
- if (doMerge) {
- if (sourceSegments == ramSegmentInfos) {
- segmentInfos.addElement(newSegment);
- } else {
- for (int i = end-1; i > minSegment; i--) // remove old infos & add new
- sourceSegments.remove(i);
+ for (int i = end-1; i > minSegment; i--) // remove old infos & add new
+ segmentInfos.remove(i);
- segmentInfos.set(minSegment, newSegment);
- }
- }
+ segmentInfos.set(minSegment, newSegment);
- if (sourceSegments == ramSegmentInfos) {
- maybeApplyDeletes(doMerge);
- doAfterFlush();
- }
-
checkpoint();
success = true;
} finally {
- if (success) {
- // The non-ram-segments case is already committed
- // (above), so all the remains for ram segments case
- // is to clear the ram segments:
- if (sourceSegments == ramSegmentInfos) {
- ramSegmentInfos.removeAllElements();
- }
- } else {
+ if (!success && rollback != null) {
+ // Rollback the individual SegmentInfo
+ // instances, but keep original SegmentInfos
+ // instance (so we don't try to write again the
+ // same segments_N file -- write once):
+ segmentInfos.clear();
+ segmentInfos.addAll(rollback);
- // Must rollback so our state matches index:
- if (sourceSegments == ramSegmentInfos && !anyDeletes) {
- // Simple case: newSegment may or may not have
- // been added to the end of our segment infos,
- // so just check & remove if so:
- if (newSegment != null &&
- segmentInfos.size() > 0 &&
- segmentInfos.info(segmentInfos.size()-1) == newSegment) {
- segmentInfos.remove(segmentInfos.size()-1);
- }
- } else if (rollback != null) {
- // Rollback the individual SegmentInfo
- // instances, but keep original SegmentInfos
- // instance (so we don't try to write again the
- // same segments_N file -- write once):
- segmentInfos.clear();
- segmentInfos.addAll(rollback);
- }
-
// Delete any partially created and now unreferenced files:
deleter.refresh();
}
}
} finally {
// close readers before we attempt to delete now-obsolete segments
- if (doMerge) merger.closeReaders();
+ merger.closeReaders();
}
- // Delete the RAM segments
- deleter.deleteDirect(ramDirectory, ramSegmentsToDelete);
-
// Give deleter a chance to remove files now.
deleter.checkpoint(segmentInfos, autoCommit);
- if (useCompoundFile && doMerge) {
+ if (useCompoundFile) {
boolean success = false;
try {
-
merger.createCompoundFile(mergedName + ".cfs");
newSegment.setUseCompoundFile(true);
checkpoint();
success = true;
-
} finally {
if (!success) {
// Must rollback:
@@ -1936,20 +2014,24 @@
deleter.checkpoint(segmentInfos, autoCommit);
}
+ long t1 = System.currentTimeMillis();
+ netMergeTime += (t1-t0);
+ if (infoStream != null)
+ System.out.println("TIME: merge: " + (netMergeTime/1000.0) + " sec");
return mergedDocCount;
}
// Called during flush to apply any buffered deletes. If
// doMerge is true then a new segment was just created and
// flushed from the ram segments.
- private final void maybeApplyDeletes(boolean doMerge) throws CorruptIndexException, IOException {
+ private final void maybeApplyDeletes(boolean flushedNewSegment) throws CorruptIndexException, IOException {
if (bufferedDeleteTerms.size() > 0) {
if (infoStream != null)
infoStream.println("flush " + numBufferedDeleteTerms + " buffered deleted terms on "
+ segmentInfos.size() + " segments.");
- if (doMerge) {
+ if (flushedNewSegment) {
IndexReader reader = null;
try {
reader = SegmentReader.get(segmentInfos.info(segmentInfos.size() - 1));
@@ -1970,7 +2052,7 @@
}
int infosEnd = segmentInfos.size();
- if (doMerge) {
+ if (flushedNewSegment) {
infosEnd--;
}
@@ -2002,6 +2084,8 @@
private final boolean checkNonDecreasingLevels(int start) {
int lowerBound = -1;
int upperBound = minMergeDocs;
+ if (upperBound == 0)
+ upperBound = 10;
for (int i = segmentInfos.size() - 1; i >= start; i--) {
int docCount = segmentInfos.info(i).docCount;
@@ -2050,10 +2134,11 @@
// well as the disk segments.
private void bufferDeleteTerm(Term term) {
Num num = (Num) bufferedDeleteTerms.get(term);
+ int numDoc = docWriter.docID;
if (num == null) {
- bufferedDeleteTerms.put(term, new Num(ramSegmentInfos.size()));
+ bufferedDeleteTerms.put(term, new Num(numDoc));
} else {
- num.setNum(ramSegmentInfos.size());
+ num.setNum(numDoc);
}
numBufferedDeleteTerms++;
}
@@ -2063,17 +2148,20 @@
// the documents buffered before it, not those buffered after it.
private final void applyDeletesSelectively(HashMap deleteTerms,
IndexReader reader) throws CorruptIndexException, IOException {
+ //System.out.println("now apply selective deletes");
Iterator iter = deleteTerms.entrySet().iterator();
while (iter.hasNext()) {
Entry entry = (Entry) iter.next();
Term term = (Term) entry.getKey();
-
+ //System.out.println(" term " + term);
+
TermDocs docs = reader.termDocs(term);
if (docs != null) {
int num = ((Num) entry.getValue()).getNum();
try {
while (docs.next()) {
int doc = docs.doc();
+ //System.out.println(" doc " + doc + " vs " + num);
if (doc >= num) {
break;
}
Index: src/java/org/apache/lucene/index/IndexFileDeleter.java
===================================================================
--- src/java/org/apache/lucene/index/IndexFileDeleter.java (revision 533617)
+++ src/java/org/apache/lucene/index/IndexFileDeleter.java (working copy)
@@ -97,6 +97,7 @@
private PrintStream infoStream;
private Directory directory;
private IndexDeletionPolicy policy;
+ private MultiDocumentWriter docWriter;
void setInfoStream(PrintStream infoStream) {
this.infoStream = infoStream;
@@ -116,10 +117,12 @@
* @throws CorruptIndexException if the index is corrupt
* @throws IOException if there is a low-level IO error
*/
- public IndexFileDeleter(Directory directory, IndexDeletionPolicy policy, SegmentInfos segmentInfos, PrintStream infoStream)
+ public IndexFileDeleter(Directory directory, IndexDeletionPolicy policy, SegmentInfos segmentInfos, PrintStream infoStream, MultiDocumentWriter docWriter)
throws CorruptIndexException, IOException {
+ this.docWriter = docWriter;
this.infoStream = infoStream;
+
this.policy = policy;
this.directory = directory;
@@ -310,6 +313,8 @@
// Incref the files:
incRef(segmentInfos, isCommit);
+ if (docWriter != null)
+ incRef(docWriter.files());
if (isCommit) {
// Append to our commits list:
@@ -325,9 +330,8 @@
// DecRef old files from the last checkpoint, if any:
int size = lastFiles.size();
if (size > 0) {
- for(int i=0;i freeThreadStates.size()) {
+ // System.out.println(" mark pending & wait..." + numThreadState + " vs " + freeThreadStates.size());
+ flushPending = true;
+ while (numThreadState > freeThreadStates.size()) {
+ // System.out.println("flush wait: " + numThreadState + " vs " + freeThreadStates.size());
+ try {
+ wait();
+ } catch (InterruptedException e) {
+ }
+ }
+ }
+
+ // nocommit: what if we hit exception before notifyAll?
+
+ fieldInfos.write(directory, segment + ".fnm");
+
+ ThreadState state = (ThreadState) freeThreadStates.get(0);
+
+ state.flushTermsAndNorms(docID);
+
+ assert fieldInfos.hasVectors() == (tvx != null);
+
+ if (tvx != null) {
+ flushedVectors = true;
+ close(tvx, tvf, tvd);
+ tvx = null;
+ } else
+ flushedVectors = false;
+
+ if (fieldsWriter != null) {
+ fieldsWriter.close();
+ fieldsWriter = null;
+ }
+
+ final int size = freeThreadStates.size();
+ for(int i=0;i 0) {
+ if (tvx == null) {
+
+ if (startByteBlock2 == null)
+ initByteBlocks2();
+
+ tvx = directory.createOutput(segment + TermVectorsWriter.TVX_EXTENSION);
+ tvx.writeInt(TermVectorsWriter.FORMAT_VERSION);
+ tvd = directory.createOutput(segment + TermVectorsWriter.TVD_EXTENSION);
+ tvd.writeInt(TermVectorsWriter.FORMAT_VERSION);
+ tvf = directory.createOutput(segment + TermVectorsWriter.TVF_EXTENSION);
+ tvf.writeInt(TermVectorsWriter.FORMAT_VERSION);
+ files = null;
+
+ // TODO: need unit test to catch this:
+ // Catch up for all previous docs null term vectors:
+ for(int i=0;i= hi)
+ return;
+
+ int mid = (lo + hi) / 2;
+
+ if (postings[lo].compareTo(postings[mid]) > 0) {
+ Posting tmp = postings[lo];
+ postings[lo] = postings[mid];
+ postings[mid] = tmp;
+ }
+
+ if (postings[mid].compareTo(postings[hi]) > 0) {
+ Posting tmp = postings[mid];
+ postings[mid] = postings[hi];
+ postings[hi] = tmp;
+
+ if (postings[lo].compareTo(postings[mid]) > 0) {
+ Posting tmp2 = postings[lo];
+ postings[lo] = postings[mid];
+ postings[mid] = tmp2;
+ }
+ }
+
+ int left = lo + 1;
+ int right = hi - 1;
+
+ if (left >= right)
+ return;
+
+ Posting partition = postings[mid];
+
+ for (; ;) {
+ while (postings[right].compareTo(partition) > 0)
+ --right;
+
+ while (left < right && postings[left].compareTo(partition) <= 0)
+ ++left;
+
+ if (left < right) {
+ Posting tmp = postings[left];
+ postings[left] = postings[right];
+ postings[right] = tmp;
+ --right;
+ } else {
+ break;
+ }
+ }
+
+ quickSort(postings, lo, left);
+ quickSort(postings, left + 1, hi);
+ }
+
+ private final void quickSort(PostingVector[] postings, int lo, int hi) {
+ if (lo >= hi)
+ return;
+
+ int mid = (lo + hi) / 2;
+
+ if (postings[lo].p.compareTo(postings[mid].p) > 0) {
+ PostingVector tmp = postings[lo];
+ postings[lo] = postings[mid];
+ postings[mid] = tmp;
+ }
+
+ if (postings[mid].p.compareTo(postings[hi].p) > 0) {
+ PostingVector tmp = postings[mid];
+ postings[mid] = postings[hi];
+ postings[hi] = tmp;
+
+ if (postings[lo].p.compareTo(postings[mid].p) > 0) {
+ PostingVector tmp2 = postings[lo];
+ postings[lo] = postings[mid];
+ postings[mid] = tmp2;
+ }
+ }
+
+ int left = lo + 1;
+ int right = hi - 1;
+
+ if (left >= right)
+ return;
+
+ PostingVector partition = postings[mid];
+
+ for (; ;) {
+ while (postings[right].p.compareTo(partition.p) > 0)
+ --right;
+
+ while (left < right && postings[left].p.compareTo(partition.p) <= 0)
+ ++left;
+
+ if (left < right) {
+ PostingVector tmp = postings[left];
+ postings[left] = postings[right];
+ postings[right] = tmp;
+ --right;
+ } else {
+ break;
+ }
+ }
+
+ quickSort(postings, lo, left);
+ quickSort(postings, left + 1, hi);
+ }
+
+ // Tokenizes the fields of a document into Postings.
+ final void processDocument(Analyzer analyzer)
+ throws IOException {
+
+ long t0 = System.currentTimeMillis();
+
+ // System.out.println("process doc");
+
+ final int numFields = numFieldData;
+
+ fdtLocal.writeVInt(numStoredFields);
+
+ if (tvx != null)
+ // TODO: really we only need to sort the subset of
+ // fields that have vectors enabled so this is
+ // wasting [not too much] time
+ Arrays.sort(fieldDataArray, 0, numFields);
+
+ // We process the document one field at a time
+ for(int i=0;i len) {
+ s.getChars(upto, upto+len, c, off);
+ upto += len;
+ left -= len;
+ return len;
+ } else if (0 == left) {
+ return -1;
+ } else {
+ s.getChars(upto, upto+left, c, off);
+ int r = left;
+ left = 0;
+ upto = s.length();
+ return r;
+ }
+ }
+ public void close() {};
+ }
+
+ ReusableStringReader stringReader = new ReusableStringReader();
+
+ // CharBlock is used to store the text of all terms. We
+ // share large arrays and terminate the text with 0xffff
+ // reserved Unicode character. We allocate needed space
+ // from the end of the block and allocate a new block
+ // whenever we've used up the current one. When we
+ // flush the Postings we then reset these blocks &
+ // repeat.
+ private final class CharBlock {
+ char[] buffer;
+ CharBlock next;
+ }
+
+ CharBlock charBlockStart;
+ CharBlock currentCharBlock;
+ char[] currentCharBuffer;
+ int currentCharUpto;
+ int currentCharLimit;
+
+ void initCharBlocks() {
+ currentCharBlock = charBlockStart = new CharBlock();
+ // TODO: tune
+ currentCharLimit = 65536;
+ currentCharBuffer = currentCharBlock.buffer = new char[currentCharLimit];
+ currentCharUpto = 0;
+ }
+
+ // Called when we have filled up the current char block
+ void newCharBlock() {
+ if (currentCharBlock.next == null) {
+ CharBlock newBlock = new CharBlock();
+ currentCharBlock.next = newBlock;
+ newBlock.next = null;
+ newBlock.buffer = new char[65536];
+ currentCharBlock = newBlock;
+ } else
+ currentCharBlock = currentCharBlock.next;
+ currentCharBuffer = currentCharBlock.buffer;
+ currentCharUpto = 0;
+ currentCharLimit = currentCharBuffer.length;
+ }
+
+ void resetCharBlocks() {
+ currentCharBlock = charBlockStart;
+ currentCharBuffer = currentCharBlock.buffer;
+ currentCharUpto = 0;
+ currentCharLimit = currentCharBuffer.length;
+ }
+
+ private final class ByteBlock {
+ byte[] buffer;
+ ByteBlock next;
+ int offset;
+
+ ByteBlock(ByteBlock prev, long startSize) {
+ if (prev != null) {
+ final int size;
+ // We can't go over 1 MB (we encode offsets into
+ // this buffer into 20 bits for term vectors):
+ if (prev.buffer.length < 1048576)
+ size = 2*prev.buffer.length;
+ else
+ size = 1048576;
+ buffer = new byte[size];
+ //System.out.println("B " + size);
+ prev.next = this;
+ offset = prev.offset+prev.buffer.length;
+ } else {
+ offset = 0;
+ buffer = new byte[(int) startSize];
+ // System.out.println("B " + startSize);
+ }
+ }
+ }
+
+ ByteBlock startByteBlock;
+ ByteBlock currentByteBlock;
+ byte[] currentByteBlockBuffer;
+ int currentByteBlockUpto;
+ int currentByteBlockOffset;
+ int currentByteBlockLimit;
+
+ long byteBlocksBytesUsed;
+
+ private void initByteBlocks() {
+ startByteBlock = currentByteBlock = new ByteBlock(null, 262144);
+ resetByteBlocks();
+ }
+
+ private void resetByteBlocks() {
+ ByteBlock b = startByteBlock;
+ while(true) {
+ Arrays.fill(b.buffer, (byte) 0);
+ if (b == currentByteBlock)
+ break;
+ b = b.next;
+ }
+ currentByteBlock = startByteBlock;
+ currentByteBlockBuffer = currentByteBlock.buffer;
+ currentByteBlockUpto = 0;
+ currentByteBlockOffset = 0;
+ currentByteBlockLimit = currentByteBlockBuffer.length;
+ byteBlocksBytesUsed = 0;
+ }
+
+ private void nextByteBlock() {
+ // Advance to next byte block
+ if (currentByteBlock.next == null)
+ // Allocate another buffer
+ new ByteBlock(currentByteBlock, 262144);
+ byteBlocksBytesUsed += currentByteBlock.buffer.length;
+ currentByteBlock = currentByteBlock.next;
+ currentByteBlockBuffer = currentByteBlock.buffer;
+ currentByteBlockUpto = 0;
+ currentByteBlockOffset = currentByteBlock.offset;
+ currentByteBlockLimit = currentByteBlockBuffer.length;
+ }
+
+ ByteBlock startByteBlock2;
+ ByteBlock currentByteBlock2;
+ byte[] currentByteBlockBuffer2;
+ int currentByteBlockUpto2;
+ int currentByteBlockOffset2;
+ int currentByteBlockLimit2;
+
+ long byteBlocksBytesUsed2;
+
+ private void initByteBlocks2() {
+ // nocommit
+ startByteBlock2 = currentByteBlock2 = new ByteBlock(null, 32768);
+ resetByteBlocks2();
+ }
+
+ private void resetByteBlocks2() {
+ ByteBlock b = startByteBlock2;
+ while(true) {
+ Arrays.fill(b.buffer, (byte) 0);
+ if (b == currentByteBlock2)
+ break;
+ b = b.next;
+ }
+ currentByteBlock2 = startByteBlock2;
+ currentByteBlockBuffer2 = currentByteBlock2.buffer;
+ currentByteBlockUpto2 = 0;
+ currentByteBlockOffset2 = 0;
+ currentByteBlockLimit2 = currentByteBlockBuffer2.length;
+ byteBlocksBytesUsed2 = 0;
+ }
+
+ private void nextByteBlock2() {
+ // Advance to next byte block
+ if (currentByteBlock2.next == null)
+ // Allocate another buffer
+ new ByteBlock(currentByteBlock2, 32768);
+ byteBlocksBytesUsed2 += currentByteBlock2.buffer.length;
+ currentByteBlock2 = currentByteBlock2.next;
+ currentByteBlockBuffer2 = currentByteBlock2.buffer;
+ currentByteBlockUpto2 = 0;
+ currentByteBlockOffset2 = currentByteBlock2.offset;
+ currentByteBlockLimit2 = currentByteBlockBuffer2.length;
+ }
+
+ // Total # Posting instances, across all fields
+ int allNumPostings;
+
+ // nocommit: must review actual # bytes per posting
+ // nocommit: in 64 bit jvm must set POINTER_NUM_BYTE=8
+ private long postingsBytesUsed() {
+ return byteBlocksBytesUsed + currentByteBlockUpto +
+ allNumPostings*(POSTING_NUM_BYTE + 2*POINTER_NUM_BYTE);
+ }
+
+ // Used to track postings for a single term. One of these
+ // exists per unique term seen since the last flush.
+ final class Posting { // info about a Term in a doc
+
+ char[] text; // holds term text (terminated by 0xffff)
+ int textStart; // offset into text where our text is stored
+ int hashCode;
+ int docFreq; // # times this term occurs in the current doc
+ int docCode; // code of prior doc
+ int numDoc; // # docs that have this term
+
+ int freqStart; // Location of first byte[] slice
+ byte[] freq; // Current byte[] buffer
+ int freqUpto; // Next write location in current byte[] slice
+
+ int proxStart; // Location of first byte[] slice
+ byte[] prox; // Current byte[] buffer
+ int proxUpto; // Next write location in current byte[] slice
+
+ int lastDocID;
+ int lastPosition;
+
+ PostingVector vector;
+
+ // ONLY USE FOR DEBUGGING! nocommit: COMMENT THIS OUT
+ public String getText() {
+ int upto = textStart;
+ while(text[upto] != 0xffff)
+ upto++;
+ return new String(text, textStart, upto-textStart);
+ }
+
+ boolean equals(String otherText) {
+ final int len = otherText.length();
+ int pos = textStart;
+ int i=0;
+ for(;i>>= 7;
+ }
+ writeFreqByte((byte) i);
+ }
+
+ public void writeProxVInt(int i) {
+ int upto = 0;
+ while ((i & ~0x7F) != 0) {
+ writeProxByte((byte)((i & 0x7f) | 0x80));
+ i >>>= 7;
+ }
+ writeProxByte((byte) i);
+ }
+
+ // Currently only used to copy a payload into the prox stream
+ public void writeProxBytes(byte[] b, int offset, int len) {
+ // TODO: we could be a bit faster here since on
+ // nextProxSlice we in fact know the # bytes we
+ // could bulk copy after that point. Though,
+ // payloads are likely to be smallish so this gain
+ // is likely not substantial.
+ final int offsetEnd = offset + len;
+ while(offset < offsetEnd) {
+ if (prox[proxUpto] != 0)
+ // End marker
+ nextProxSlice();
+ prox[proxUpto++] = b[offset++];
+ }
+ }
+
+ // Called when we have filled up the current freq
+ // slice and we need to write another byte
+ public void nextFreqSlice() {
+
+ int newSize;
+ int newLevel;
+
+ switch(freq[freqUpto]&7) {
+ case 0:
+ newLevel = 1;
+ newSize = 10;
+ break;
+ case 1:
+ newLevel = 2;
+ newSize = 10;
+ break;
+ case 2:
+ newLevel = 3;
+ newSize = 20;
+ break;
+ case 3:
+ newLevel = 4;
+ newSize = 20;
+ break;
+ case 4:
+ newLevel = 5;
+ newSize = 40;
+ break;
+ case 5:
+ newLevel = 6;
+ newSize = 80;
+ break;
+ default:
+ newLevel = 7;
+ newSize = 160;
+ break;
+ }
+
+ if (currentByteBlockLimit - currentByteBlockUpto < newSize)
+ nextByteBlock();
+
+ final int upto = currentByteBlockUpto;
+ final int offset = upto + currentByteBlockOffset;
+ currentByteBlockUpto += newSize;
+
+ // Copy forward the past 3 bytes (which we are about
+ // to overwrite with the forwarding address):
+ currentByteBlockBuffer[upto] = freq[freqUpto-3];
+ currentByteBlockBuffer[upto+1] = freq[freqUpto-2];
+ currentByteBlockBuffer[upto+2] = freq[freqUpto-1];
+
+ // Write forwarding address at end of last slice:
+ freq[freqUpto-3] = (byte) (offset >>> 24);
+ freq[freqUpto-2] = (byte) (offset >>> 16);
+ freq[freqUpto-1] = (byte) (offset >>> 8);
+ freq[freqUpto] = (byte) offset;
+
+ // Switch to new slice
+ freq = currentByteBlockBuffer;
+ freqUpto = upto+3;
+ freq[currentByteBlockUpto-1] = (byte) (8 + newLevel);
+ }
+
+ // Called when we have used up our current prox slice
+ // and we need to write another byte
+ public void nextProxSlice() {
+
+ int newSize;
+ int newLevel;
+
+ switch(prox[proxUpto]&7) {
+ case 0:
+ newLevel = 1;
+ newSize = 10;
+ break;
+ case 1:
+ newLevel = 2;
+ newSize = 10;
+ break;
+ case 2:
+ newLevel = 3;
+ newSize = 20;
+ break;
+ case 3:
+ newLevel = 4;
+ newSize = 20;
+ break;
+ case 4:
+ newLevel = 5;
+ newSize = 40;
+ break;
+ case 5:
+ newLevel = 6;
+ newSize = 80;
+ break;
+ default:
+ newLevel = 7;
+ newSize = 160;
+ break;
+ }
+
+ if (currentByteBlockLimit - currentByteBlockUpto < newSize)
+ nextByteBlock();
+
+ if (vector != null && 0 == (vector.positionStartData & 1)) {
+
+ // Maybe go back and record the end point of this
+ // first slice
+ final int start = vector.positionStartData >>> 12;
+
+ if (proxUpto <= (3+start)) {
+
+ // No bytes were written for this document to
+ // this prox slice (the last 3 bytes are moved
+ // to the start of the next slice), so, record
+ // on the next prox slice instead:
+ vector.positionSlice = currentByteBlockBuffer;
+ vector.positionStartData = ((currentByteBlockUpto + 3 - (proxUpto - start)) << 12);
+ //System.out.println(" term=" + getText() + " tv: postpone proxStart start=" + currentByteBlockUpto);
+
+ } else {
+
+ // 3 bits for level
+ final int level = prox[proxUpto]&7;
+
+ // 8 bits for fragment
+ final int fragment = proxUpto-start-3;
+
+ assert fragment < 256;
+
+ // 20 bits for upto
+ assert proxUpto < 1048576;
+ assert start < 1048576;
+
+ //System.out.println(" term=" + getText() + " tv: set proxStart start=" + start + " level=" + level + " fragment=" + fragment + " nextIndex=" + (currentByteBlockUpto + currentByteBlockOffset));
+
+ // Encodes start position, start level, #bytes left:
+ vector.positionStartData = (start<<12) | (fragment<<4) | (level<<1) | 1;
+ }
+ }
+
+ final int upto = currentByteBlockUpto;
+ final int offset = upto + currentByteBlockOffset;
+ currentByteBlockUpto += newSize;
+
+ // Copy forward the past 3 bytes (which we are about
+ // to overwrite with the forwarding address):
+ currentByteBlockBuffer[upto] = prox[proxUpto-3];
+ currentByteBlockBuffer[upto+1] = prox[proxUpto-2];
+ currentByteBlockBuffer[upto+2] = prox[proxUpto-1];
+
+ // Write forwarding address at end of last slice:
+ prox[proxUpto-3] = (byte) (offset >>> 24);
+ prox[proxUpto-2] = (byte) (offset >>> 16);
+ prox[proxUpto-1] = (byte) (offset >>> 8);
+ prox[proxUpto] = (byte) offset;
+
+ // Switch to new slice
+ prox = currentByteBlockBuffer;
+ proxUpto = upto+3;
+
+ // Mark the end of this slice
+ prox[currentByteBlockUpto-1] = (byte) (8 + newLevel);
+ }
+
+ public void writeProxByte(byte b) {
+ if (prox[proxUpto] != 0)
+ nextProxSlice();
+ prox[proxUpto++] = b;
+ }
+
+ public void writeFreqByte(byte b) {
+ if (freq[freqUpto] != 0)
+ nextFreqSlice();
+ freq[freqUpto++] = b;
+ }
+
+ // Flush all output, given start slice, to the
+ // provided IndexOutput stream
+ public long writeTo(IndexOutput out, int start, byte[] bufferEnd, int uptoEnd) throws IOException {
+
+ // TODO: binary search?
+ // Find our starting block
+ ByteBlock block = startByteBlock;
+ byte[] buffer = block.buffer;
+ ByteBlock next = block.next;
+ while(block.offset + buffer.length <= start) {
+ block = next;
+ buffer = block.buffer;
+ next = next.next;
+ }
+
+ int level = 0;
+ int upto = start-block.offset;
+ int limit;
+
+ if (buffer == bufferEnd && uptoEnd < upto+5)
+ // Only one slice:
+ limit = uptoEnd;
+ else
+ limit = upto + 1;
+
+ long size = 0;
+
+ // ASSERT
+ int lastLoc = start;
+
+ while(true) {
+ if (buffer == bufferEnd && uptoEnd <= limit) {
+ // Last slice
+ out.writeBytes(buffer, upto, uptoEnd-upto);
+ size += uptoEnd-upto;
+ break;
+ } else {
+ // Not the last slice
+ out.writeBytes(buffer, upto, limit-upto);
+ size += limit-upto;
+
+ // Move to next slice
+ final int newSize;
+
+ switch(level) {
+ case 0:
+ level = 1;
+ newSize = 10;
+ break;
+ case 1:
+ level = 2;
+ newSize = 10;
+ break;
+ case 2:
+ level = 3;
+ newSize = 20;
+ break;
+ case 3:
+ level = 4;
+ newSize = 20;
+ break;
+ case 4:
+ level = 5;
+ newSize = 40;
+ break;
+ case 5:
+ level = 6;
+ newSize = 80;
+ break;
+ default:
+ level = 7;
+ newSize = 160;
+ break;
+ }
+
+ // Seek to the next buffer
+ final int nextLoc = ((buffer[limit]&0xff)<<24) + ((buffer[1+limit]&0xff)<<16) + ((buffer[2+limit]&0xff)<<8) + (buffer[3+limit]&0xff);
+
+ // ASSERT
+ assert nextLoc > lastLoc;
+ lastLoc = nextLoc;
+
+ // Maybe advance buffers
+ while(block.offset + buffer.length <= nextLoc) {
+ block = next;
+ buffer = block.buffer;
+ next = next.next;
+ }
+
+ upto = nextLoc - block.offset;
+ if (buffer == bufferEnd && uptoEnd < upto+newSize)
+ // Last slice
+ limit = uptoEnd;
+ else
+ // Not the last slice
+ limit = upto + newSize - 4;
+ }
+ }
+
+ return size;
+ }
+ }
+
+ // Used to track data for term vectors. One of these
+ // exists per unique term seen in the document. We
+ // tap into the positions storage in the Posting, but
+ // for offsets we use our own byte array.
+ final class PostingVector {
+
+ Posting p;
+ int lastOffset;
+
+ // For storing offsets
+ int sliceStart;
+ byte[] slice;
+ int upto;
+
+ // For referencing positions
+ byte[] positionSlice;
+ int positionStartData; // encodes data (upto, level, limit) needed to read encoded positions
+
+ public void initSlice() {
+ final int newSize = 5;
+ if (currentByteBlockLimit2 - currentByteBlockUpto2 < newSize)
+ nextByteBlock2();
+
+ slice = currentByteBlockBuffer2;
+ upto = currentByteBlockUpto2;
+ currentByteBlockUpto2 += newSize;
+ slice[currentByteBlockUpto2-1] = 8;
+ sliceStart = upto + currentByteBlock2.offset;
+ }
+
+ public void writeVInt(int i) {
+ int upto = 0;
+ while ((i & ~0x7F) != 0) {
+ writeByte((byte)((i & 0x7f) | 0x80));
+ i >>>= 7;
+ }
+ writeByte((byte) i);
+ }
+
+ public void writeByte(byte b) {
+
+ // System.out.println(" writeByte b=" + b + " upto=" + upto + " limit=" + limit + " buffer=" + slice);
+
+ if (slice[upto] != 0) {
+
+ int newSize;
+ int newLevel;
+
+ // Determine size & level of next slice
+ switch(slice[upto]&7) {
+ case 0:
+ newLevel = 1;
+ newSize = 10;
+ break;
+ case 1:
+ newLevel = 2;
+ newSize = 10;
+ break;
+ case 2:
+ newLevel = 3;
+ newSize = 20;
+ break;
+ case 3:
+ newLevel = 4;
+ newSize = 20;
+ break;
+ case 4:
+ newLevel = 5;
+ newSize = 40;
+ break;
+ case 5:
+ newLevel = 6;
+ newSize = 80;
+ break;
+ default:
+ newLevel = 7;
+ newSize = 160;
+ break;
+ }
+
+ // Maybe allocate another block
+ if (currentByteBlockLimit2 - currentByteBlockUpto2 < newSize)
+ nextByteBlock2();
+
+ final int newUpto = currentByteBlockUpto2;
+ final int offset = newUpto + currentByteBlock2.offset;
+ currentByteBlockUpto2 += newSize;
+
+ // Copy forward the past 3 bytes (which we are about
+ // to overwrite with the forwarding address):
+ currentByteBlockBuffer2[newUpto] = slice[upto-3];
+ currentByteBlockBuffer2[newUpto+1] = slice[upto-2];
+ currentByteBlockBuffer2[newUpto+2] = slice[upto-1];
+
+ // Write forwarding address at end of last slice:
+ slice[upto-3] = (byte) (offset >>> 24);
+ slice[upto-2] = (byte) (offset >>> 16);
+ slice[upto-1] = (byte) (offset >>> 8);
+ slice[upto] = (byte) offset;
+
+ // Switch to new slice
+ slice = currentByteBlockBuffer2;
+ upto = newUpto+3;
+
+ // Write new level:
+ slice[currentByteBlockUpto2-1] = (byte) (8 + newLevel);
+ }
+ slice[upto++] = b;
+ }
+
+ // Flush all output we've accumulated to the IndexOutput
+ public void flush(RAMWriter out) {
+
+ // Find our starting block
+ ByteBlock block = startByteBlock2;
+ byte[] slice0 = block.buffer;
+ ByteBlock next = block.next;
+ while(block.offset + slice0.length <= sliceStart) {
+ block = next;
+ slice0 = block.buffer;
+ next = next.next;
+ }
+ int level0 = 0;
+ int upto0 = sliceStart-block.offset;
+
+ int limit0;
+ if (slice0 == slice && upto < upto0+5)
+ // Only one slice
+ limit0 = upto;
+ else
+ limit0 = upto0+1;
+
+ while(true) {
+ if (slice == slice0 && upto > upto0 && upto <= limit0) {
+ // This is the end
+ out.writeBytes(slice, upto0, upto-upto0);
+ break;
+ } else {
+ out.writeBytes(slice0, upto0, limit0-upto0);
+
+ // Move to next slice
+ final int newSize;
+
+ switch(level0) {
+ case 0:
+ level0 = 1;
+ newSize = 10;
+ break;
+ case 1:
+ level0 = 2;
+ newSize = 10;
+ break;
+ case 2:
+ level0 = 3;
+ newSize = 20;
+ break;
+ case 3:
+ level0 = 4;
+ newSize = 20;
+ break;
+ case 4:
+ level0 = 5;
+ newSize = 40;
+ break;
+ case 5:
+ level0 = 6;
+ newSize = 80;
+ break;
+ default:
+ level0 = 7;
+ newSize = 160;
+ break;
+ }
+
+ // Seek to the next slice
+ final int nextLoc = ((slice0[limit0]&0xff)<<24) + ((slice0[1+limit0]&0xff)<<16) + ((slice0[2+limit0]&0xff)<<8) + (slice0[3+limit0]&0xff);
+
+ while(block.offset + slice0.length <= nextLoc) {
+ block = next;
+ slice0 = block.buffer;
+ next = next.next;
+ }
+
+ upto0 = nextLoc - block.offset;
+
+ if (slice0 == slice && upto < upto0+newSize)
+ // Last slice
+ limit0 = upto;
+ else
+ limit0 = upto0 + newSize - 4;
+ }
+ }
+ }
+ }
+
+ char[] localTextBuffer = new char[10];
+
+ Posting p;
+
+ PostingVector[] postingsVectorsArray = new PostingVector[10];
+ int postingsVectorsUpto;
+
+ private RAMSegment ramSegment;
+
+ private final class ByteSliceReader extends IndexInput {
+ ByteBlock next;
+ ByteBlock block;
+ byte[] buffer;
+ int upto;
+ int limit;
+ int level;
+
+ byte[] bufferEnd;
+ int uptoEnd;
+
+ // ASSERT
+ int lastSlice;
+
+ public void init(ByteBlock startBlock, int startIndex, byte[] bufferEnd, int uptoEnd) {
+ // TODO: we could do binary search
+ // Seek to the starting block
+
+ this.uptoEnd = uptoEnd;
+ this.bufferEnd = bufferEnd;
+
+ block = startBlock;
+ buffer = block.buffer;
+ next = block.next;
+ while(block.offset + buffer.length <= startIndex) {
+ block = next;
+ buffer = block.buffer;
+ next = next.next;
+ }
+ level = 0;
+ upto = startIndex-block.offset;
+
+ // ASSERT
+ lastSlice = startIndex;
+
+ final int firstSize = 5;
+
+ if (buffer == bufferEnd && upto+firstSize >= uptoEnd) {
+ // There is only this one slice to read
+ assert upto + firstSize - uptoEnd > 0;
+ limit = uptoEnd;
+ } else
+ limit = upto+firstSize-4;
+ }
+
+ // Used to initialize partway through a slice:
+ public void init(ByteBlock startBlock, byte[] startBuffer, int loc, int startBytesLeft, int startLevel, byte[] bufferEnd, int uptoEnd) {
+
+ this.uptoEnd = uptoEnd;
+ this.bufferEnd = bufferEnd;
+
+ // Seek to the starting block
+ block = startBlock;
+ next = block.next;
+
+ // ASSERT
+ int totOffset = 0;
+
+ // TODO: we could do binary search
+ while(block.buffer != startBuffer) {
+ // ASSERT
+ totOffset += block.buffer.length;
+
+ block = next;
+ next = next.next;
+ }
+ level = startLevel;
+ upto = loc;
+ buffer = startBuffer;
+
+ // ASSERT
+ lastSlice = totOffset+loc;
+
+ limit = upto+startBytesLeft;
+ }
+
+ public byte readByte() {
+ if (upto == limit)
+ nextSlice();
+ return buffer[upto++];
+ }
+
+ public long writeTo(IndexOutput out) throws IOException {
+ long size = 0;
+ while(true) {
+ if (buffer == bufferEnd && uptoEnd == limit) {
+ assert uptoEnd >= upto;
+ out.writeBytes(buffer, upto, uptoEnd-upto);
+ size += uptoEnd-upto;
+ break;
+ } else {
+ out.writeBytes(buffer, upto, limit-upto);
+ size += limit-upto;
+ nextSlice();
+ }
+ }
+
+ return size;
+ }
+
+ public void nextSlice() {
+
+ // Skip to our next slice
+ final int nextIndex = ((buffer[limit]&0xff)<<24) + ((buffer[1+limit]&0xff)<<16) + ((buffer[2+limit]&0xff)<<8) + (buffer[3+limit]&0xff);
+
+ // ASSERT
+ assert nextIndex > lastSlice;
+ lastSlice = nextIndex;
+
+ final int newSize;
+
+ switch(level) {
+ case 0:
+ level = 1;
+ newSize = 10;
+ break;
+ case 1:
+ level = 2;
+ newSize = 10;
+ break;
+ case 2:
+ level = 3;
+ newSize = 20;
+ break;
+ case 3:
+ level = 4;
+ newSize = 20;
+ break;
+ case 4:
+ level = 5;
+ newSize = 40;
+ break;
+ case 5:
+ level = 6;
+ newSize = 80;
+ break;
+ default:
+ level = 7;
+ newSize = 160;
+ break;
+ }
+
+ while(block.offset + buffer.length <= nextIndex) {
+ block = next;
+ buffer = block.buffer;
+ next = next.next;
+ }
+
+ upto = nextIndex-block.offset;
+
+ if (buffer == bufferEnd && upto+newSize >= uptoEnd) {
+ // We are advancing to the final slice
+ assert upto+newSize-uptoEnd > 0;
+ // nocommit -- is this OK?
+ //limit = upto+newSize;
+ limit = uptoEnd;
+ } else {
+ // This is not the final slice (subtract 4 for the
+ // forwarding address at the end of this new slice)
+ limit = upto+newSize-4;
+ }
+ }
+
+ public void readBytes(byte[] b, int offset, int len) {
+ while(len > 0) {
+ final int numLeft = limit-upto;
+ if (numLeft < len) {
+ // Read entire slice
+ System.arraycopy(buffer, upto, b, offset, numLeft);
+ offset += numLeft;
+ len -= numLeft;
+ nextSlice();
+ } else {
+ // This slice is the last one
+ System.arraycopy(buffer, upto, b, offset, len);
+ upto += len;
+ break;
+ }
+ }
+ }
+
+ public long getFilePointer() {throw new RuntimeException("not implemented");}
+ public long length() {throw new RuntimeException("not implemented");}
+ public void seek(long pos) {throw new RuntimeException("not implemented");}
+ public void close() {throw new RuntimeException("not implemented");}
+ }
+
+ ByteSliceReader byteSliceReader = new ByteSliceReader();
+ ByteSliceReader byteSliceReader3 = new ByteSliceReader();
+ ByteSliceReader byteSliceReader2 = new ByteSliceReader();
+
+ int maxTermLen;
+ int postingsDocCount;
+
+ public void buildRAMSegment() throws IOException {
+ long ta = System.currentTimeMillis();
+
+ assert termsOut.buffer == null;
+ assert freqOut.buffer == null;
+ assert proxOut.buffer == null;
+
+ if (infoStream != null) {
+ infoStream.println("\n" + getElapsedTime() + ": build ram segment docID=" + docID);
+ System.out.println(" RAM: " + nf.format(getRAMUsed()/1024./1024.) + " MB");
+ }
+
+ long oldSize = postingsBytesUsed();
+
+ termsOut.setStartLevel(4);
+ freqOut.setStartLevel(4);
+ proxOut.setStartLevel(4);
+
+ final int numFields = numAllFieldData;
+
+ Arrays.sort(allFieldDataArray, 0, numFields);
+ for(int i=0;i ramBufferSize/14) {
+ //if (levelCounts[0] > 1500) {
+ long t0 = System.currentTimeMillis();
+ mergeRAMSegments(this, 0);
+ netMerge0Time += (System.currentTimeMillis()-t0);
+ if (levelSizes[1] > ramBufferSize/7 && level0Compression < 0.7) {
+ t0 = System.currentTimeMillis();
+ mergeRAMSegments(this, 1);
+ netMerge1Time += (System.currentTimeMillis()-t0);
+ }
+ }
+ */
+ }
+
+ /*
+ private final void addNorms() throws IOException {
+ for(int n=0;n0) position += analyzer.getPositionIncrementGap(fieldInfo.name);
+
+ if (!field.isTokenized()) { // un-tokenized field
+ token = localToken;
+ String stringValue = field.stringValue();
+ token.setTermText(stringValue);
+ token.setStartOffset(offset);
+ token.setEndOffset(offset + stringValue.length());
+ addPosition();
+ offset += stringValue.length();
+ length++;
+ } else { // tokenized field
+ TokenStream stream = field.tokenStreamValue();
+
+ // the field does not have a TokenStream,
+ // so we have to obtain one from the analyzer
+ if (stream == null) {
+ final Reader reader; // find or make Reader
+ if (field.readerValue() != null)
+ reader = field.readerValue();
+ else {
+ stringReader.init(field.stringValue());
+ reader = stringReader;
+ }
+
+ // Tokenize field and add to postingTable
+ stream = analyzer.tokenStream(fieldInfo.name, reader);
+ }
+
+ // reset the TokenStream to the first token
+ stream.reset();
+
+ try {
+ offsetEnd = offset-1;
+ for (token = stream.next(); token != null; token = stream.next()) {
+ position += (token.getPositionIncrement() - 1);
+ addPosition();
+ if (++length >= maxFieldLength) {
+ if (infoStream != null)
+ infoStream.println("maxFieldLength " +maxFieldLength+ " reached, ignoring following tokens");
+ break;
+ }
+ }
+ offset = offsetEnd+1;
+ } finally {
+ stream.close();
+ }
+ }
+
+ boost *= field.getBoost();
+ }
+
+
+
+ /*
+ * Walk through all unique text tokens (Posting
+ * instances) found in this field and serialize them
+ * into a single RAM segment.
+ */
+ void addPostings()
+ throws CorruptIndexException, IOException {
+
+ final int numTerms = numPostings;
+
+ //System.out.println(" addPostings field " + fieldInfo.name + ": " + numTerms + " unique terms seen");
+
+ // Write "field changed" marker:
+ if (numTerms > 0) {
+ //System.out.println(" write fieldChanged fieldNumber=" + fieldInfo.number + " buffer=" + termsOut.buffer+ " upto=" + termsOut.upto);
+ termsOut.writeVInt(END_MARKER);
+ termsOut.writeVInt(fieldInfo.number);
+ }
+
+ final Posting[] postings = postingsArray;
+ quickSort(postings, 0, numTerms-1);
+
+ Posting lastPosting = null;
+
+ long lastFreqPos = 0;
+ long lastProxPos = 0;
+
+ for(int i=0;i one");
+ } else {
+ left = posting.proxUpto - start;
+ level = 0;
+ //System.out.println(" one");
+ }
+
+ //System.out.println(" write positions start=" + start + " left=" + left + " level=" + level + " startBuf=" + startByteBlock.buffer + " slice=" + vector.positionSlice + " prox=" + posting.prox + " proxUpto=" + posting.proxUpto);
+
+ byteSliceReader.init(startByteBlock,
+ vector.positionSlice,
+ start, left, level,
+ posting.prox, posting.proxUpto);
+
+ // We can't do blind copy because format is not
+ // the same (due to payloads). We can change
+ // prox format to be identical when there are no
+ // payloads to get back to this straight copy;
+ // it's just somewhat more complex.
+
+ // byteSliceReader.writeTo(tvfLocal, posting.prox, posting.proxUpto);
+
+ for(int i=0;i>>1);
+ if ((code & 1) != 0) {
+ final int len = byteSliceReader.readVInt();
+ // TODO: more efficient "skip bytes"
+ for(int k=0;k textBuffer.length)
+ textBuffer = new char[(int) (maxTermLen*1.5)];
+ }
+
+ public long flushProx(IndexOutput out) throws IOException {
+ copyBytes(prox, out, proxSize);
+ return proxSize;
+ }
+
+ public long flushFreq(IndexOutput out, int lastDocID) throws IOException {
+ // We only need to "stitch up" the boundary by
+ // re-encoding our first docID as a delta to the
+ // lastDocID of the previous freq
+ final int docCode = freq.readVInt();
+ assert 0==lastDocID || (docCode >>> 1) > lastDocID;
+ final int newDocCode = (((docCode >>> 1) - lastDocID) << 1) | (docCode & 1);
+ out.writeVInt(newDocCode);
+ final int sz = vIntSize(docCode);
+ if (freqSize>sz)
+ copyBytes(freq, out, freqSize - sz);
+ return freqSize + vIntSize(newDocCode) - sz;
+ }
+
+ public int sort(int[] result) {
+ int num = 0;
+ SegmentMergeInfo smi2 = this;
+ while(smi2 != null) {
+ result[num++] = smi2.idx;
+ smi2 = smi2.next;
+ }
+
+ if (2 == num) {
+ if (result[0] > result[1]) {
+ final int t = result[0];
+ result[0] = result[1];
+ result[1] = t;
+ }
+ } else
+ // TODO: maybe radix sort here?
+ Arrays.sort(result, 0, num);
+
+ return num;
+ }
+
+ public boolean next() throws IOException {
+
+ int start = terms.readVInt();
+
+ if (start == END_MARKER) {
+ fieldNumber = terms.readVInt();
+ if (fieldNumber == END_MARKER)
+ return false;
+ else
+ // Field changed
+ start = terms.readVInt();
+ }
+
+ assert start <= textLength;
+ final int length = terms.readVInt();
+ textLength = start + length;
+ assert textLength <= textBuffer.length;
+
+ // TODO: we could readChars & compute hash code in 1 loop
+ terms.readChars(textBuffer, start, length);
+
+ hashCode = 0;
+ for(int i=textLength-1;i>=0;i--)
+ hashCode = (hashCode * 37) + textBuffer[i];
+
+ numDoc = terms.readVInt();
+ lastDocID = terms.readVInt();
+
+ freqSize = terms.readVLong();
+ proxSize = terms.readVLong();
+
+ return true;
+ }
+
+ public void close() throws IOException {
+ MultiDocumentWriter.close(terms, freq, prox);
+ }
+
+ public boolean equals(SegmentMergeInfo other) {
+ if (other.fieldNumber == fieldNumber &&
+ other.textLength == textLength) {
+ final char[] textA = textBuffer;
+ final char[] textB = other.textBuffer;
+ for(int i=0;i>> 1) > lastDocID;
+ final int newDocCode = (((docCode >>> 1) - lastDocID) << 1) | (docCode & 1);
+ out.writeVInt(newDocCode);
+ final long sz = byteSliceReader.writeTo(out);
+ return sz + vIntSize(newDocCode);
+ }
+
+ public boolean next() {
+
+ if (postingsLimit == postingsUpto) {
+
+ // See if we should advance to the next field...
+
+ while(true) {
+ if (fieldIndex == numAllFieldData)
+ // We hit the last field, so we are done
+ return false;
+
+ FieldData fp = allFieldDataArray[fieldIndex++];
+ postingsLimit = fp.numPostings;
+ if (postingsLimit > 0) {
+ if (infoStream != null)
+ System.out.println(" field " + fp.fieldInfo.name + ": " + postingsLimit + " unique terms seen");
+ postingsArray = fp.postingsArray;
+ postingsHash = fp.postingsHash;
+ postingsHashMask = fp.postingsHashMask;
+ fieldNumber = fp.fieldInfo.number;
+ postingsUpto = 0;
+ quickSort(postingsArray, 0, postingsLimit-1);
+ break;
+ }
+ }
+ }
+
+ p = postingsArray[postingsUpto++];
+
+ // Write last entry in freq
+ if (1 == p.docFreq)
+ p.writeFreqVInt(p.docCode|1);
+ else {
+ p.writeFreqVInt(p.docCode);
+ p.writeFreqVInt(p.docFreq);
+ }
+
+ lastDocID = p.lastDocID;
+ numDoc = p.numDoc;
+
+ int offset = p.textStart;
+ final char[] text = p.text;
+ while(text[offset] != 0xffff)
+ offset++;
+ textLength = offset - p.textStart;
+
+ // TODO: we could avoid this copy by overloading
+ // compare
+
+ if (textLength > maxTermLen)
+ maxTermLen = textLength;
+ if (textLength > textBuffer.length)
+ textBuffer = new char[(int) (textLength*1.5)];
+
+ System.arraycopy(text, p.textStart, textBuffer, 0, textLength);
+ hashCode = p.hashCode;
+
+ byteSliceReader.init(startByteBlock, p.freqStart, p.freq, p.freqUpto);
+ byteSliceReader3.init(startByteBlock, p.proxStart, p.prox, p.proxUpto);
+
+ return true;
+ }
+ }
+
+ /*
+ This queue is used for merging RAM and Flushed
+ segments. It's modified from the PriorityQueue used
+ for main segment merging: it has two tiers. The first
+ tier, using a priority queue, keeps track of each
+ unique term that's we've seen. The second tier, using
+ linked list inside SMI, keeps track of all SMIs that
+ have this term. This "de-dupping" is a good
+ performance gain when you are merging a very large
+ number of segments since the "lessThan" method is
+ quite costly.
+ */
+
+ // Shared merge queue
+ MergeQueue mergeQueue = new MergeQueue();
+
+ final class MergeQueue {
+
+ // Records all idx's that are pending for a given field+text:
+ private SegmentMergeInfo[] heap;
+ private SegmentMergeInfo[] hash;
+ private int size;
+ private int maxSize;
+ private int hashMask;
+
+ void init(int newMaxSize) {
+ size = 0;
+ if (maxSize < newMaxSize) {
+ if (newMaxSize < 32)
+ maxSize = 32;
+ else
+ maxSize = (int) (1.25*newMaxSize);
+ int heapSize = maxSize + 1;
+ heap = new SegmentMergeInfo[heapSize];
+ this.maxSize = maxSize;
+ int hashSize = 32;
+ int target = 3*maxSize;
+ while(hashSize < target)
+ hashSize *= 2;
+ hash = new SegmentMergeInfo[hashSize];
+ hashMask = hashSize-1;
+ }
+ }
+
+ /**
+ * Adds a SegmentMergeInfo to a PriorityQueue in log(size) time.
+ * If one tries to add more objects than maxSize from initialize
+ * a RuntimeException (ArrayIndexOutOfBound) is thrown.
+ */
+ public void put(SegmentMergeInfo smi) {
+
+ //System.out.println("Q: put text=" + new String(smi.textBuffer, 0, smi.textLength) + " field=" + smi.fieldNumber + " idx=" + smi.idx + " smi=" + smi + " hash=" + smi.hashCode);
+
+ // See if the term for this SMI is already hashed
+ int hashPos = smi.hashCode & hashMask;
+ SegmentMergeInfo smi2 = hash[hashPos];
+ //System.out.println(" hash[" + hashPos + "] = " + smi2);
+ while(smi2 != null && (smi2.hashCode != smi.hashCode || !smi.equals(smi2)))
+ smi2 = smi2.hashNext;
+
+ if (smi2 != null) {
+ // This term is already in the queue, so we don't
+ // add it again. Instead, we chain it (linked
+ // list) to the SMI already enrolled.
+ smi.next = smi2.next;
+ smi2.next = smi;
+ // System.out.println(" already seen");
+ } else {
+ // First time we are seeing this field+text, so
+ // enroll into hash & priority queue:
+ heap[++size] = smi;
+ smi.next = null;
+ smi.hashNext = hash[hashPos];
+ hash[hashPos] = smi;
+ upHeap();
+ // System.out.println(" not yet seen; set hash[" + hashPos + "]=" + smi + "; set smi.hashNext=" + smi.hashNext);
+ }
+ }
+
+ /** Removes and returns the least element of the PriorityQueue in log(size)
+ time. */
+ public SegmentMergeInfo pop() {
+ SegmentMergeInfo smi = heap[1]; // save first value
+ // System.out.println("Q: pop text=" + new String(smi.textBuffer, 0, smi.textLength));
+ heap[1] = heap[size]; // move last to first
+ size--;
+ downHeap(); // adjust heap
+
+ // Also remove from hash:
+ int hashPos = smi.hashCode & hashMask;
+ SegmentMergeInfo lastSmi2 = null;
+ SegmentMergeInfo smi2 = hash[hashPos];
+ while(smi2 != smi) {
+ lastSmi2 = smi2;
+ smi2 = smi2.hashNext;
+ }
+ assert smi2 != null;
+ if (lastSmi2 == null)
+ hash[hashPos] = smi.hashNext;
+ else
+ lastSmi2.hashNext = smi.hashNext;
+ return smi;
+ }
+
+ private void upHeap() {
+ int i = size;
+ SegmentMergeInfo node = heap[i]; // save bottom node
+ int j = i >>> 1;
+ while (j > 0 && lessThan(node, heap[j])) {
+ heap[i] = heap[j]; // shift parents down
+ i = j;
+ j = j >>> 1;
+ }
+ heap[i] = node; // install saved node
+ }
+
+ private void downHeap() {
+ int i = 1;
+ SegmentMergeInfo node = heap[i]; // save top node
+ int j = i << 1; // find smaller child
+ int k = j + 1;
+ if (k <= size && lessThan(heap[k], heap[j])) {
+ j = k;
+ }
+ while (j <= size && lessThan(heap[j], node)) {
+ heap[i] = heap[j]; // shift up child
+ i = j;
+ j = i << 1;
+ k = j + 1;
+ if (k <= size && lessThan(heap[k], heap[j])) {
+ j = k;
+ }
+ }
+ heap[i] = node; // install saved node
+ }
+
+ // return true if a < b
+ protected boolean lessThan(SegmentMergeInfo stiA, SegmentMergeInfo stiB) {
+
+ // first by field
+ if (stiA.fieldNumber == stiB.fieldNumber) {
+
+ // then by text
+
+ // TODO: most of the time we are comparing things
+ // with long shared prefixes; is there some way to
+ // optimize for this fact?
+ final char[] textA = stiA.textBuffer;
+ final char[] textB = stiB.textBuffer;
+ final int len = stiA.textLength < stiB.textLength ? stiA.textLength : stiB.textLength;
+ for(int i=0;i charB)
+ return false;
+ }
+
+ if (stiA.textLength < stiB.textLength)
+ return true;
+ else if (stiA.textLength > stiB.textLength)
+ return false;
+
+ // Should never get here because dups are handled by
+ // first tier hash:
+ //System.out.println(" failed text=" + new String(stiA.textBuffer, 0, stiA.textLength));
+ assert false;
+ return false;
+
+ } else {
+ // fields differ:
+ String fieldA = fieldInfos.fieldName(stiA.fieldNumber);
+ String fieldB = fieldInfos.fieldName(stiB.fieldNumber);
+ return fieldA.compareTo(fieldB) < 0;
+ }
+ }
+ }
+
+ //int lastDocID;
+ char[] lastChars = new char[10];
+
+ // Merges RAM segments into a single segment, which may be
+ // in RAM or in the real directory. Input segments for
+ // merging should be placed in mergeInputs already.
+ final void mergeTerms(int numSegmentsIn, IndexOutput termsOut, IndexOutput freqOut, IndexOutput proxOut,
+ boolean includePostingsHash) throws IOException {
+
+ MergeQueue queue = null;
+
+ // nocommit
+ //boolean debug = false;
+
+ queue = mergeQueue;
+ queue.init(1+numSegmentsIn);
+
+ final SegmentMergeInfo[] inputs = mergeInputs;
+
+ // initialize queue
+ for (int i=0;i 0) {
+ fp.resetPostingArrays();
+ fp.numPostings = 0;
+ }
+ }
+ }
+
+ private void initAllPostingArrays() {
+ for(int i=0;i 0) {
+ fp.initPostingArrays();
+ fp.numPostings = 0;
+ }
+ }
+ }
+
+ private int vIntSize(int v) {
+ int count = 1;
+ while ((v & ~0x7F) != 0) {
+ count++;
+ v >>>= 7;
+ }
+ return count;
+ }
+
+ private final TermInfo termInfo = new TermInfo(); // minimize consing
+ private IndexOutput freqOutput;
+ private IndexOutput proxOutput;
+ private int skipInterval;
+ private int lastDoc;
+ private int lastPayloadLength;
+ private int df;
+ private boolean currentFieldStorePayloads;
+
+ // Write out the postings & dictionary to real output
+ // files, in the "real" lucene file format. This is to
+ // finalize a segment.
+ void flushTermsAndNorms(int totalNumDoc) throws IOException {
+
+ if (infoStream != null)
+ infoStream.println("\nflush postings as segment " + segment + " docID=" + MultiDocumentWriter.this.docID);
+
+ // First we must pre-merge flushed segments:
+ /*
+ boolean any;
+ do {
+ any = false;
+ for(int i=flushedLevelCounts.length-1;i>=0;i--)
+ if (flushedLevelCounts[i] > 0) {
+ // Merge up all levels below the current max level:
+ for(int j=0;j 0) {
+ mergeFlushedSegments(this, j);
+ any = true;
+ }
+
+ // Do one more merge if we have too many flushed
+ // segments at the max level:
+ if (flushedLevelCounts[i] > flushedMergeFactor) {
+ mergeFlushedSegments(this, i);
+ any = true;
+ }
+ break;
+ }
+ } while(any);
+ */
+ while(flushedSegments.size() > flushedMergeFactor) {
+ if (infoStream != null)
+ infoStream.println(" merge flushed segments before flushing terms: now " + flushedSegments.size() + " flushed segments");
+ mergeFlushedSegments(this, flushedSegments.size()-flushedMergeFactor, flushedSegments.size(), -1);
+ }
+
+ if (infoStream != null)
+ infoStream.println("now create segment " + segment);
+
+ TermInfosWriter termInfosWriter = null;
+
+ final int numRAMSegments = ramSegments.size();
+ final int numFlushedSegments = flushedSegments.size();
+ final int numSegmentsIn = numRAMSegments + numFlushedSegments;
+
+ resizeMergeInputs(numSegmentsIn);
+
+ int numDoc = 0;
+ long oldSize = 0;
+ long newSize = 0;
+
+ final SegmentMergeInfo[] inputs = mergeInputs;
+
+ try {
+ freqOutput = directory.createOutput(segment + ".frq");
+ proxOutput = directory.createOutput(segment + ".prx");
+ termInfosWriter = new TermInfosWriter(directory, segment, fieldInfos,
+ writer.getTermIndexInterval());
+ skipInterval = termInfosWriter.skipInterval;
+
+ MergeQueue queue = mergeQueue;
+ queue.init(1+numSegmentsIn);
+
+ int i=0;
+ for (;i 0) {
+ if (REUSE_BYTE_ARRAYS)
+ resetByteBlocks();
+ else
+ initByteBlocks();
+
+ if (REUSE_CHAR_ARRAYS)
+ resetCharBlocks();
+ else
+ initCharBlocks();
+
+ if (!REUSE_POSTING_ARRAYS)
+ initAllPostingArrays();
+ else
+ resetAllPostingArrays();
+
+ postingsDocCount = 0;
+ allNumPostings = 0;
+ }
+
+ // NOTE: if we merged flushed segments, we have now just
+ // obsoleted some files. But we don't call deleter
+ // checkpoint here because our caller (IndexWriter) will
+ // do so shortly after calling us.
+ }
+
+ /* Called only by flushTerms, to append all postings for
+ * a given term into the main freq/prox postings
+ * output. */
+ void appendPostings(SegmentMergeInfo smi) throws IOException {
+
+ final IndexInput freq = smi.freq;
+ final IndexInput prox = smi.prox;
+
+ int doc = 0;
+ boolean done = false;
+ int numDoc = 0;
+
+ while (numDoc++ < smi.numDoc) {
+
+ if ((++df % skipInterval) == 0)
+ bufferSkip(lastDoc, lastPayloadLength);
+
+ final int docCode = freq.readVInt();
+ doc += docCode >>> 1;
+ assert doc <= maxDocID;
+ assert doc > lastDoc || df == 1;
+
+ final int termDocFreq;
+
+ if ((docCode&1) != 0)
+ termDocFreq = 1;
+ else
+ termDocFreq = freq.readVInt();
+
+ final int newDocCode = (doc-lastDoc)<<1;
+ lastDoc = doc;
+
+ // Carefully copy over the prox + payload info,
+ // changing the format to match Lucene's segment
+ // format.
+ for(int i=0;i>1);
+ }
+ }
+
+ // We can't do this more efficient "raw copy" below
+ // due to payloads; if we change prox format to be
+ // identical when there are no payloads then we can
+ // go back to this. It's just somewhate more
+ // complex but I believe feasible:
+
+ /*
+ // Copy prox's
+ int count = 0;
+ while(count < termDocFreq) {
+ byte b = prox.readByte();
+ count += (b & 128) == 0 ? 1:0;
+ proxOutput.writeByte(b);
+ }
+ */
+
+ // TODO: we can speed this up by not actually
+ // interp'ing the vints
+ if (1 == termDocFreq) {
+ freqOutput.writeVInt(newDocCode|1);
+ } else {
+ freqOutput.writeVInt(newDocCode);
+ freqOutput.writeVInt(termDocFreq);
+ }
+ }
+ }
+
+ private RAMWriter skipBuffer = new RAMWriter();
+ private int lastSkipDoc;
+ private int lastSkipPayloadLength;
+ private long lastSkipFreqPointer;
+ private long lastSkipProxPointer;
+
+ private void resetSkip() {
+ lastSkipDoc = 0;
+ lastSkipPayloadLength = -1; // we don't have to write the first length in the skip list
+ lastSkipFreqPointer = freqOutput.getFilePointer();
+ lastSkipProxPointer = proxOutput.getFilePointer();
+ }
+
+ private void bufferSkip(int doc, int payloadLength) throws IOException {
+ //System.out.println(" buffer skip: freq ptr " + freqPointer + " prox " + proxPointer);
+ //System.out.println(" vs last freq ptr " + lastSkipFreqPointer + " prox " + lastSkipProxPointer);
+
+ // To efficiently store payloads in the posting lists we do not store the length of
+ // every payload. Instead we omit the length for a payload if the previous payload had
+ // the same length.
+ // However, in order to support skipping the payload length at every skip point must be known.
+ // So we use the same length encoding that we use for the posting lists for the skip data as well:
+ // Case 1: current field does not store payloads
+ // SkipDatum --> DocSkip, FreqSkip, ProxSkip
+ // DocSkip,FreqSkip,ProxSkip --> VInt
+ // DocSkip records the document number before every SkipInterval th document in TermFreqs.
+ // Document numbers are represented as differences from the previous value in the sequence.
+ // Case 2: current field stores payloads
+ // SkipDatum --> DocSkip, PayloadLength?, FreqSkip,ProxSkip
+ // DocSkip,FreqSkip,ProxSkip --> VInt
+ // PayloadLength --> VInt
+ // In this case DocSkip/2 is the difference between
+ // the current and the previous value. If DocSkip
+ // is odd, then a PayloadLength encoded as VInt follows,
+ // if DocSkip is even, then it is assumed that the
+ // current payload length equals the length at the previous
+ // skip point
+
+ final int delta = doc - lastSkipDoc;
+ if (currentFieldStorePayloads) {
+ if (payloadLength == lastSkipPayloadLength)
+ // the current payload length equals the length at the previous skip point,
+ // so we don't store the length again
+ skipBuffer.writeVInt(delta << 1);
+ else {
+ // the payload length is different from the previous one. We shift the DocSkip,
+ // set the lowest bit and store the current payload length as VInt.
+ skipBuffer.writeVInt((delta << 1) + 1);
+ skipBuffer.writeVInt(payloadLength);
+ lastSkipPayloadLength = payloadLength;
+ }
+ } else
+ // current field does not store payloads
+ skipBuffer.writeVInt(delta);
+
+ long freqPointer = freqOutput.getFilePointer();
+ long proxPointer = proxOutput.getFilePointer();
+ skipBuffer.writeVInt((int) (freqPointer - lastSkipFreqPointer));
+ skipBuffer.writeVInt((int) (proxPointer - lastSkipProxPointer));
+ lastSkipFreqPointer = freqPointer;
+ lastSkipProxPointer = proxPointer;
+
+ lastSkipDoc = doc;
+ }
+
+ long writeSkip() throws IOException {
+ long skipPointer = freqOutput.getFilePointer();
+ skipBuffer.writeTo(freqOutput);
+ return skipPointer;
+ }
+
+ SegmentMergeInfo mergeInputs[] = new SegmentMergeInfo[0];
+ SegmentMergeInfo segmentMergeInfos[] = new SegmentMergeInfo[0];
+ PostingsHashMergeInfo postingsMergeInfo = new PostingsHashMergeInfo();
+ int[] mergeIDXArray;
+
+ final void resizeMergeInputs(final int minSize) {
+ // Must to 1+ to allow for PostingsHashMergeInfo
+ if (mergeInputs.length < 1+minSize) {
+ int size = (int) ((1+minSize)*1.25);
+ SegmentMergeInfo[] newArray = new SegmentMergeInfo[size];
+ System.arraycopy(segmentMergeInfos, 0, newArray, 0, segmentMergeInfos.length);
+ for(int i=segmentMergeInfos.length;i= MAX_WAIT_QUEUE) {
+ // System.out.println("do wait");
+
+ // There are too many thread states in line write
+ // to the index so we now pause to give them a
+ // chance to get scheduled by the JVM and finish
+ // their documents. Once we wake up again, a
+ // recycled ThreadState should be available else
+ // we wait again.
+ // System.out.println("w " + Thread.currentThread().getName());
+ try {
+ wait();
+ } catch (InterruptedException e) {
+ }
+ // System.out.println(" wd " + Thread.currentThread().getName());
+
+ } else {
+ // OK, just create a new thread state
+ state = new ThreadState();
+ numThreadState++;
+ break;
+ }
+ } else {
+ // Use recycled thread state
+ state = (ThreadState) freeThreadStates.get(size-1);
+ freeThreadStates.remove(size-1);
+ break;
+ }
+ }
+
+ boolean success = false;
+ try {
+ state.init(doc, docID++);
+ success = true;
+ } finally {
+ if (!success)
+ freeThreadStates.add(state);
+ }
+
+ return state;
+ }
+
+ void addDocument(Document doc, Analyzer analyzer)
+ throws CorruptIndexException, IOException {
+
+ // System.out.println("\nadd doc docID=" + docID);
+
+ // First pass: go through all fields in doc, updating
+ // shared FieldInfos and writing any stored fields:
+ final ThreadState state = getThreadState(doc);
+ boolean success = false;
+ try {
+ state.processDocument(analyzer);
+ success = true;
+ } finally {
+ if (success)
+ finishDocument(state);
+ else {
+ // nocommit: need to do some cleanup of the thread state?
+ freeThreadStates.add(state);
+ }
+ }
+ // System.out.println(" done");
+ }
+
+ long netMerge0Time;
+ long netMerge1Time;
+ long netFlushedMergeTime;
+ long netDocTime;
+ long netProcessTime;
+ long netFlushTime;
+ long netSegmentTime;
+ int maxTermLen;
+
+ /*
+ * Does the synchronized work to finish/flush the inverted document.
+ */
+ private synchronized void finishDocument(ThreadState state) throws IOException {
+
+ maxTermLen = state.maxTermLen > maxTermLen ? state.maxTermLen : maxTermLen;
+
+ // Now write the indexed document to the real files.
+
+ // THREADS: only 1 thread now so this must be the case:
+ assert nextWriteDocID == state.docID;
+
+ if (nextWriteDocID == state.docID) {
+ // It's my turn, so write everything now:
+ try {
+ writeDocument(state);
+ } finally {
+ nextWriteDocID++;
+ // Recycle our thread state back in the free pool
+ freeThreadStates.add(state);
+ }
+
+ // If any states were waiting on me, sweep through and
+ // flush those that are enabled by my write.
+ boolean doNotify = numWaiting >= MAX_WAIT_QUEUE || flushPending;
+ if (numWaiting > 0) {
+ while(true) {
+ int upto = 0;
+ for(int i=0;i 0) {
+ tvd.writeVInt(state.numVectorFields);
+ for(int i=0;i ramBufferSize/3) {
+ // Time to flush the postings hash
+ if (postingsSize + totalRAMSegmentSize> ramBufferSize) {
+ // Time to flush to disk
+ if (doSelfFlush) {
+ // Flush to a flushed segment
+ flushRAMSegments(state);
+ } else {
+ // We do nothing here because writer will call
+ // flush...
+ }
+ } else {
+ // OK just build a RAM segment
+ System.out.println("BEFORE: " + nf.format(getRAMUsed()/1024./1024.));
+ state.buildRAMSegment();
+ System.out.println("AFTER: " + nf.format(getRAMUsed()/1024./1024.));
+ printTimes();
+ }
+ }
+ }
+
+ // For debugging only
+ /*
+ public void printAlloc(String prefix, RAMCell head, int limit) {
+ RAMCell c = head;
+ System.out.print(prefix + ":");
+ if (c == null)
+ System.out.println(" null");
+ else
+ while(c != null) {
+ if (c.next == null) {
+ System.out.println(" " + c.buffer.length + "(" + limit + ")");
+ break;
+ } else {
+ System.out.print(" " + c.buffer.length);
+ c = c.next;
+ }
+ }
+ }
+ */
+
+ long getRAMUsed() {
+ return totalRAMSegmentSize + totalPostingsSize;
+ }
+
+ private final String tempFileName(int count, String suffix) {
+ // nocommit suffix should come from IndexFileNames
+ return segment + "x" + count + "." + suffix;
+ }
+
+ // Called when RAM buffer is full; we now merge all RAM
+ // segments to a single flushed segment:
+ final synchronized void flushRAMSegments(ThreadState state) throws IOException {
+
+ long t0 = System.currentTimeMillis();
+
+ // THREADS: for thread concurrency we must step through
+ // all ThreadStates here
+ if (infoStream != null) {
+ String name = tempFileName(flushedCount, ".txx");
+ infoStream.println("\n" + getElapsedTime() + ": flush ram segments at docID " + docID + ", to " + name.substring(0, name.length()-4) + ": totalRam=" + nf.format(totalRAMSegmentSize/1024./1024.) + " MB");
+ }
+ System.out.println("FLUSH TEMP @ docID=" + docID + " numDoc=" + (docID-lastFlushDocID) + "; RAM=" + totalRAMSegmentSize);
+ System.out.println(" mem now: " + bean.getHeapMemoryUsage().getUsed());
+ lastFlushDocID = docID;
+
+ IndexOutput termsOut = directory.createOutput(tempFileName(flushedCount, "txx"));
+ IndexOutput freqOut = directory.createOutput(tempFileName(flushedCount, "fxx"));
+ IndexOutput proxOut = directory.createOutput(tempFileName(flushedCount, "pxx"));
+
+ final int numSegmentsIn = ramSegments.size();
+ long newSize;
+ long oldSize;
+ if (state.postingsDocCount > 0)
+ oldSize = getRAMUsed();
+ else
+ oldSize = totalRAMSegmentSize;
+
+ state.resizeMergeInputs(numSegmentsIn);
+
+ int numDoc = 0;
+ for(int i=0;i=0;i--)
+ System.out.println(" level " + i + ": count=" + flushedLevelCounts[i]);
+
+ files = null;
+
+ // Must "register" our newly created files with the
+ // deleter so that when they later decref they get
+ // deleted:
+ synchronized(writer) {
+ writer.getDeleter().checkpoint(writer.segmentInfos, false);
+ }
+ netFlushTime += (System.currentTimeMillis()-t0);
+
+ if (infoStream != null)
+ printTimes();
+
+ // Maybe cascade merges. We do slightly different
+ // policy than normal segment merges: we let 20 level 0
+ // segments accumulate first, then we merge the first 10
+ // into a level 1 segment. After another 10 level 0
+ // segments we merge the first 10 level 0's into another
+ // level 1, etc. This better "spreads" / "postpones"
+ // the merge work so we don't pay a massive wasted merge
+ // price only to find it's time to flush a real segment.
+ int mergeLevel = 0;
+ while(mergeLevel < flushedLevelCounts.length && flushedLevelCounts[mergeLevel] == 2*flushedMergeFactor)
+ mergeFlushedSegments(state, mergeLevel++);
+ }
+
+ // Merge flushed segments into a single new flushed segment
+ final void mergeFlushedSegments(ThreadState state, int level) throws IOException {
+
+ int start = 0;
+ int end = 0;
+ for(int i=flushedLevelCounts.length-1;i>=level;i--) {
+ start = end;
+ end += flushedLevelCounts[i];
+ }
+
+ if (end-start > flushedMergeFactor)
+ end = start+flushedMergeFactor;
+
+ if (infoStream != null)
+ infoStream.println("merge flushed segments: level " + level);
+
+ mergeFlushedSegments(state, start, end, level);
+ }
+
+ final void mergeFlushedSegments(ThreadState state, int start, int end, int level) throws IOException {
+ long t0 = System.currentTimeMillis();
+ if (infoStream != null) {
+ String name = tempFileName(flushedCount, ".txx");
+ infoStream.println("merge flushed segments to " + name.substring(0, name.length()-4) + ": start " + start + " to end " + end);
+ }
+
+ long newSize;
+ int numDoc;
+ long oldSize;
+
+ FlushedSegment newSegment;
+
+ if (1 == end-start) {
+ // Degenerate case
+ newSegment = (FlushedSegment) flushedSegments.get(start);
+ numDoc = newSegment.numDoc;
+ oldSize = newSize = newSegment.size;
+ } else {
+
+ // maybe reallocate
+ state.resizeMergeInputs(end-start);
+ numDoc = 0;
+ oldSize = 0;
+ IndexOutput termsOut = directory.createOutput(tempFileName(flushedCount, "txx"));
+ IndexOutput freqOut = directory.createOutput(tempFileName(flushedCount, "fxx"));
+ IndexOutput proxOut = directory.createOutput(tempFileName(flushedCount, "pxx"));
+
+ try {
+ int upto = 0;
+ for (int i=start;i fieldIdx;
+ //System.out.println(" fs " + j + ": no norms; fillBytes: " + fs.numDoc + " bytes");
+ fillBytes(termsOut, defaultNorm, fs.numDoc);
+ }
+ }
+ }
+ }
+
+ // Write end marker
+ termsOut.writeVInt(END_MARKER);
+ termsOut.writeVInt(END_MARKER);
+
+ newSize = termsOut.getFilePointer() + freqOut.getFilePointer() + proxOut.getFilePointer();
+ } finally {
+ close(termsOut, freqOut, proxOut);
+ for(int i=0;i start; i--) // remove old infos & add new
+ flushedSegments.remove(i);
+
+ newSegment = new FlushedSegment(numDoc, flushedCount++, newSize);
+ flushedSegments.set(start, newSegment);
+
+ if (infoStream != null)
+ printTimes();
+ }
+
+
+ if (level != -1) {
+ if (flushedLevelSizes.length == level+1) {
+ flushedLevelSizes = realloc(flushedLevelSizes, 1+flushedLevelSizes.length);
+ flushedLevelCounts = realloc(flushedLevelCounts, 1+flushedLevelCounts.length);
+ }
+
+ flushedLevelSizes[level] -= oldSize;
+ flushedLevelSizes[1+level] += newSize;
+
+ flushedLevelCounts[level] -= (end-start);
+ flushedLevelCounts[1+level]++;
+ }
+
+ totalFlushedSize += newSize - oldSize;
+
+ if (infoStream != null) {
+ infoStream.println(" done: oldSize=" + oldSize + " newSize=" + newSize + " new/old=" + ((int)(100.0*newSize/oldSize)) + "% totalFlushed=" + (totalFlushedSize/1024/1024) + " MB");
+ if (level != -1)
+ for(int i=flushedLevelCounts.length-1;i>=0;i--)
+ System.out.println(" level " + i + ": count=" + flushedLevelCounts[i]);
+ }
+
+ files = null;
+
+ // nocommit: should I just give IFD list of files to delete?
+ // Have deleter remove our now unreferenced files:
+ synchronized(writer) {
+ writer.getDeleter().checkpoint(writer.segmentInfos, false);
+ }
+ netFlushedMergeTime += System.currentTimeMillis()-t0;
+
+ printTimes();
+ }
+
+ static void close(IndexOutput f0, IndexOutput f1, IndexOutput f2) throws IOException {
+ IOException keep = null;
+ try {
+ if (f0 != null) f0.close();
+ } catch (IOException e) {
+ keep = e;
+ } finally {
+ try {
+ if (f1 != null) f1.close();
+ } catch (IOException e) {
+ if (keep == null) keep = e;
+ } finally {
+ try {
+ if (f2 != null) f2.close();
+ } catch (IOException e) {
+ if (keep == null) keep = e;
+ } finally {
+ if (keep != null) throw keep;
+ }
+ }
+ }
+ }
+
+ static void close(IndexInput f0, IndexInput f1, IndexInput f2) throws IOException {
+ IOException keep = null;
+ try {
+ if (f0 != null) f0.close();
+ } catch (IOException e) {
+ keep = e;
+ } finally {
+ try {
+ if (f1 != null) f1.close();
+ } catch (IOException e) {
+ if (keep == null) keep = e;
+ } finally {
+ try {
+ if (f2 != null) f2.close();
+ } catch (IOException e) {
+ if (keep == null) keep = e;
+ } finally {
+ if (keep != null) throw keep;
+ }
+ }
+ }
+ }
+
+ static void close(IndexOutput freq, IndexOutput prox, TermInfosWriter terms) throws IOException {
+ IOException keep = null;
+ try {
+ if (freq != null) freq.close();
+ } catch (IOException e) {
+ keep = e;
+ } finally {
+ try {
+ if (prox != null) prox.close();
+ } catch (IOException e) {
+ if (keep == null) keep = e;
+ } finally {
+ try {
+ if (terms != null) terms.close();
+ } catch (IOException e) {
+ if (keep == null) keep = e;
+ } finally {
+ if (keep != null) throw keep;
+ }
+ }
+ }
+ }
+
+ float level0Compression;
+
+ NumberFormat nf = NumberFormat.getInstance();
+ String getElapsedTime() {
+ long t = System.currentTimeMillis();
+ nf.setMaximumFractionDigits(1);
+ nf.setMinimumFractionDigits(1);
+ return nf.format((t-startTime)/1000.0) + " sec";
+ }
+
+ // In-memory merge: reads multiple ram segments (in the
+ // modified format) and replaces with a single ram segment.
+ final void mergeRAMSegments(ThreadState state, int level) throws IOException {
+
+ int start = 0;
+ int end = 0;
+ for(int i=levelCounts.length-1;i>=level;i--) {
+ start = end;
+ end += levelCounts[i];
+ }
+
+ if (infoStream != null) {
+ infoStream.println("\n" + getElapsedTime() + ": merge ram segments: level " + level + ": start idx " + start + " to end idx " + end + " docID=" + docID);
+ System.out.println(" RAM: " + nf.format(getRAMUsed()/1024./1024.) + " MB");
+ }
+ long oldSize;
+ //long oldTermsSize;
+ //long oldFreqSize;
+ //long oldProxSize;
+ long newSize;
+
+ int numDoc;
+ RAMSegment newRAMSegment;
+
+ if (end == start+1) {
+ // Degenerate case, if suddenly an immense document
+ // comes through
+ newRAMSegment = (RAMSegment) ramSegments.get(start);
+ //oldTermsSize = newRAMSegment.terms.size;
+ //oldFreqSize = newRAMSegment.freq.size;
+ //oldProxSize = newRAMSegment.prox.size;
+ newSize = oldSize = newRAMSegment.size;
+ numDoc = newRAMSegment.numDoc;
+ } else {
+
+ state.resizeMergeInputs(end-start);
+ final int numSegmentsIn = end-start;
+
+ oldSize = 0;
+ //oldTermsSize = 0;
+ //oldFreqSize = 0;
+ //oldProxSize = 0;
+ int upto = 0;
+ numDoc = 0;
+ for(int i=start;i 0;
+
+ if (srcIn instanceof RAMReader) {
+ RAMReader src = (RAMReader) srcIn;
+ while(true) {
+ final int chunk = src.limit - src.upto;
+ if (chunk < numBytes) {
+ // Src is the limit
+ destIn.writeBytes(src.buffer, src.upto, chunk);
+ src.nextBuffer();
+ numBytes -= chunk;
+ } else if (chunk == numBytes) {
+ // Matched
+ destIn.writeBytes(src.buffer, src.upto, chunk);
+ src.nextBuffer();
+ break;
+ } else {
+ // numBytes is the limit
+ destIn.writeBytes(src.buffer, src.upto, (int) numBytes);
+ src.upto += numBytes;
+ break;
+ }
+ }
+ } else {
+ // Use intermediate buffer
+ while(numBytes > 0) {
+ final int chunk;
+ if (numBytes > 1024) {
+ chunk = 1024;
+ } else {
+ chunk = (int) numBytes;
+ }
+ srcIn.readBytes(byteBuffer, 0, chunk);
+ destIn.writeBytes(byteBuffer, 0, chunk);
+ numBytes -= chunk;
+ }
+ }
+ }
+
+ /** If non-null, a message will be printed to this if maxFieldLength is reached.
+ */
+ void setInfoStream(PrintStream infoStream) {
+ this.infoStream = infoStream;
+ // nocommit
+ //this.infoStream = System.out;
+ }
+
+ private class RAMSegment {
+ int numDoc;
+ RAMCell freq;
+ int freqLimit;
+ RAMCell prox;
+ int proxLimit;
+ RAMCell terms;
+ int termsLimit;
+ long size;
+
+ public RAMSegment(int numDoc, RAMWriter terms, RAMWriter freq, RAMWriter prox) {
+ this.numDoc = numDoc;
+
+ size = terms.size + freq.size + prox.size;
+
+ this.terms = terms.head;
+ this.termsLimit = terms.upto;
+ terms.reset();
+
+ this.freq = freq.head;
+ this.freqLimit = freq.upto;
+ freq.reset();
+
+ this.prox = prox.head;
+ this.proxLimit = prox.upto;
+ prox.reset();
+ }
+ }
+
+ private class FlushedSegment {
+ int numDoc;
+ int segment;
+ long size;
+ public FlushedSegment(int numDoc, int segment, long size) {
+ this.numDoc = numDoc;
+ this.segment = segment;
+ this.size = size;
+ }
+ }
+
+ final static int MAX_RAM_CELL_LEVEL = 4;
+ RAMCell freeCells[] = new RAMCell[1+MAX_RAM_CELL_LEVEL];
+
+ synchronized void recycle(RAMCell cell) {
+ cell.next = freeCells[cell.level];
+ freeCells[cell.level] = cell;
+ }
+
+ public RAMCell alloc(final int level, final int subLevel) {
+ RAMCell r;
+ synchronized(this) {
+ r = freeCells[level];
+ if (r != null)
+ freeCells[level] = r.next;
+ else
+ r = null;
+ }
+
+ if (r == null)
+ r = new RAMCell(level, subLevel);
+ else {
+ r.next = null;
+ r.subLevel = (byte) subLevel;
+ }
+ return r;
+ }
+
+ private static final class RAMCell {
+
+ byte[] buffer;
+ RAMCell next;
+ byte level;
+ byte subLevel;
+
+ public RAMCell(final int level, final int subLevel) {
+ this.level = (byte) level;
+ this.subLevel = (byte) subLevel;
+ int size = 0;
+ switch(this.level) {
+ case 0:
+ size = 64;
+ break;
+ case 1:
+ size = 256;
+ break;
+ case 2:
+ size = 1024;
+ break;
+ case 3:
+ size = 4096;
+ break;
+ case 4:
+ size = 16384;
+ break;
+ }
+ buffer = new byte[size];
+ //System.out.println("BC " + size);
+ }
+ }
+
+ private final class RAMWriter extends IndexOutput {
+
+ RAMCell head;
+ RAMCell tail;
+ int upto;
+ int limit;
+ byte[] buffer;
+ long size;
+
+ boolean isFree = false;
+
+ void setStartLevel(int level) {
+ assert head == null;
+ head = tail = alloc(level, 0);
+ buffer = head.buffer;
+ upto = 0;
+ limit = head.buffer.length;
+ size = limit;
+ }
+
+ public void writeByte(byte b) {
+ assert !isFree;
+ if (upto == limit)
+ nextBuffer();
+
+ buffer[upto++] = b;
+ }
+
+ public void writeChars(char[] s, int start, int length) {
+ final int end = start + length;
+ for (int i = start; i < end; i++) {
+ final int code = (int)s[i];
+ if (code >= 0x01 && code <= 0x7F)
+ writeByte((byte)code);
+ else if (((code >= 0x80) && (code <= 0x7FF)) || code == 0) {
+ writeByte((byte)(0xC0 | (code >> 6)));
+ writeByte((byte)(0x80 | (code & 0x3F)));
+ } else {
+ writeByte((byte)(0xE0 | (code >>> 12)));
+ writeByte((byte)(0x80 | ((code >> 6) & 0x3F)));
+ writeByte((byte)(0x80 | (code & 0x3F)));
+ }
+ }
+ }
+
+ public void writeVInt(int i) {
+ while ((i & ~0x7F) != 0) {
+ writeByte((byte)((i & 0x7f) | 0x80));
+ i >>>= 7;
+ }
+ writeByte((byte)i);
+ }
+
+ // Move all of our bytes to out and reset
+ public void writeTo(IndexOutput out) throws IOException {
+ assert !isFree;
+ while(head != null) {
+ final int numBytes;
+ if (head.next == null)
+ numBytes = upto;
+ else
+ numBytes = head.buffer.length;
+ out.writeBytes(head.buffer, 0, numBytes);
+ RAMCell next = head.next;
+ recycle(head);
+ head = next;
+ }
+ reset();
+ }
+
+ private void reset() {
+ assert !isFree;
+ head = tail = null;
+ buffer = null;
+ limit = upto = 0;
+ size = 0;
+ }
+
+ void free() {
+ assert !isFree;
+ while(head != null) {
+ RAMCell c = head.next;
+ recycle(head);
+ head = c;
+ }
+ reset();
+ }
+
+ RAMCell localCell;
+ int localUpto;
+
+ private void writeByteLocal(byte b) {
+ if (localUpto == localCell.buffer.length) {
+ localCell = localCell.next;
+ localUpto = 0;
+ }
+ //System.out.println(" upto=" + localUpto + " len=" + localCell.buffer.length);
+ localCell.buffer[localUpto++] = b;
+ }
+
+ // Write an int at a specific spot:
+ public void writeInt(RAMCell cell, int upto, int v) {
+ localCell = cell;
+ localUpto = upto;
+ writeByteLocal((byte) (v>>24));
+ writeByteLocal((byte) (v>>16));
+ writeByteLocal((byte) (v>>8));
+ writeByteLocal((byte) v);
+ }
+
+ public void writeVLong(RAMCell cell, int upto, long v, int fillCount) {
+ localCell = cell;
+ localUpto = upto;
+ // System.out.println("writeVLong " + v + " fillCount=" + fillCount);
+ while ((v & ~0x7F) != 0) {
+ writeByteLocal((byte)((v & 0x7f) | 0x80));
+ fillCount--;
+ v >>>= 7;
+ }
+ writeByteLocal((byte) v);
+ fillCount--;
+ while(fillCount > 0) {
+ //System.out.println(" FILL 0 buffer=" + localCell.buffer + " upto=" + upto);
+ writeByteLocal((byte)0);
+ fillCount--;
+ }
+ }
+
+ public void writeBytes(byte[] b, int offset, int numBytes) {
+ assert !isFree;
+ assert numBytes > 0;
+ switch(numBytes) {
+ case 4:
+ writeByte(b[offset++]);
+ case 3:
+ writeByte(b[offset++]);
+ case 2:
+ writeByte(b[offset++]);
+ case 1:
+ writeByte(b[offset++]);
+ break;
+ default:
+ if (upto == limit)
+ nextBuffer();
+ // System.out.println(" writeBytes: buffer=" + buffer + " head=" + head + " offset=" + offset + " nB=" + numBytes);
+ while(true) {
+ int chunk = limit - upto;
+ if (chunk >= numBytes) {
+ System.arraycopy(b, offset, buffer, upto, numBytes);
+ upto += numBytes;
+ break;
+ } else {
+ System.arraycopy(b, offset, buffer, upto, chunk);
+ offset += chunk;
+ numBytes -= chunk;
+ nextBuffer();
+ }
+ }
+ }
+ }
+
+ public void nextBuffer() {
+ assert !isFree;
+
+ final int level;
+ final int subLevel;
+ if (tail == null) {
+ level = 0;
+ subLevel = 0;
+ } else if (tail.level < MAX_RAM_CELL_LEVEL) {
+ if (7 == tail.subLevel) {
+ level = 1+tail.level;
+ subLevel = 0;
+ } else {
+ level = tail.level;
+ subLevel = 1+tail.subLevel;
+ }
+ } else {
+ subLevel = 0;
+ level = MAX_RAM_CELL_LEVEL;
+ }
+
+ RAMCell c = alloc(level, subLevel);
+
+ if (head == null)
+ head = tail = c;
+ else {
+ tail.next = c;
+ tail = c;
+ }
+
+ limit = c.buffer.length;
+ size += limit;
+ buffer = c.buffer;
+
+ upto = 0;
+ }
+
+ public long getFilePointer() {
+ assert !isFree;
+ return size - (limit-upto);
+ }
+
+ public long length() {
+ assert !isFree;
+ return getFilePointer();
+ }
+
+ public void close() {}
+ public void flush() {throw new RuntimeException("not implemented");}
+ public void seek(long pos) {throw new RuntimeException("not implemented");}
+ }
+
+ // Limited IndexInput for "read once". This frees each
+ // buffer from the head once it's been read.
+ private final class RAMReader extends IndexInput {
+
+ int readLimit;
+ int upto;
+ int limit;
+ RAMCell head;
+ byte[] buffer;
+ long pos;
+
+ // ASSERT
+ boolean finished = true;
+
+ RAMReader(RAMCell head, int limit) {
+ reset(head, limit);
+ }
+
+ public void reset(RAMCell head, int limit) {
+ // Make sure we were fully read
+ assert finished;
+ finished = false;
+ readLimit = limit;
+ this.head = head;
+ upto = 0;
+ pos = 0;
+
+ if (head == null) {
+ assert 0 == readLimit;
+ buffer = null;
+ } else {
+ buffer = head.buffer;
+ if (head.next == null) {
+ this.limit = readLimit;
+ assert this.limit > 0 && this.limit <= buffer.length;
+ } else
+ this.limit = buffer.length;
+ }
+ }
+
+ public byte readByte() {
+ byte b = buffer[upto++];
+ if (upto == limit)
+ nextBuffer();
+ return b;
+ }
+
+ public void nextBuffer() {
+ RAMCell c = head.next;
+ pos += limit;
+ recycle(head);
+ head = c;
+ upto = 0;
+ if (head != null) {
+ buffer = head.buffer;
+ if (head.next == null) {
+ limit = readLimit;
+ assert limit > 0 && limit <= buffer.length;
+ } else
+ limit = buffer.length;
+ } else {
+ // ASSERT
+ finished = true;
+ buffer = null;
+ }
+ }
+
+ public void seek(long toPos) {
+ assert toPos > (pos+upto);
+ while(true) {
+ if (pos+limit > toPos) {
+ // Seek within current buffer
+ upto = (int) (toPos-pos);
+ break;
+ } else
+ nextBuffer();
+ }
+ }
+
+ public long getFilePointer() {
+ return pos+upto;
+ }
+
+ public void readBytes(byte[] b, int offset, int len) {throw new RuntimeException("not implemented");}
+ public void close() {}
+ public long length() {throw new RuntimeException("not implemented");}
+ }
+
+ static final byte defaultNorm = Similarity.encodeNorm(1.0f);
+
+ private class BufferedNorms {
+
+ RAMWriter out = new RAMWriter();
+ int upto;
+
+ void add(float norm) {
+ byte b = Similarity.encodeNorm(norm);
+ out.writeByte(b);
+ upto++;
+ }
+
+ void fill(int docID) {
+ while(upto < docID) {
+ // fill in docs that didn't have this field:
+ out.writeByte(defaultNorm);
+ upto++;
+ }
+ }
+ }
+
+ static long[] realloc(long[] array, int newSize) {
+ long[] newArray = new long[newSize];
+ System.arraycopy(array, 0, newArray, 0, array.length);
+ return newArray;
+ }
+
+ static int[] realloc(int[] array, int newSize) {
+ int[] newArray = new int[newSize];
+ System.arraycopy(array, 0, newArray, 0, array.length);
+ return newArray;
+ }
+}
Property changes on: src/java/org/apache/lucene/index/MultiDocumentWriter.java
___________________________________________________________________
Name: svn:eol-style
+ native
Index: src/java/org/apache/lucene/index/SegmentMergeInfo.java
===================================================================
--- src/java/org/apache/lucene/index/SegmentMergeInfo.java (revision 533617)
+++ src/java/org/apache/lucene/index/SegmentMergeInfo.java (working copy)
@@ -73,9 +73,8 @@
final void close() throws IOException {
termEnum.close();
- if (postings != null) {
- postings.close();
+ if (postings != null)
+ postings.close();
}
}
-}
Index: src/java/org/apache/lucene/store/IndexOutput.java
===================================================================
--- src/java/org/apache/lucene/store/IndexOutput.java (revision 533617)
+++ src/java/org/apache/lucene/store/IndexOutput.java (working copy)
@@ -125,6 +125,24 @@
}
}
+ public void writeChars(char[] s, int start, int length)
+ throws IOException {
+ final int end = start + length;
+ for (int i = start; i < end; i++) {
+ final int code = (int)s[i];
+ if (code >= 0x01 && code <= 0x7F)
+ writeByte((byte)code);
+ else if (((code >= 0x80) && (code <= 0x7FF)) || code == 0) {
+ writeByte((byte)(0xC0 | (code >> 6)));
+ writeByte((byte)(0x80 | (code & 0x3F)));
+ } else {
+ writeByte((byte)(0xE0 | (code >>> 12)));
+ writeByte((byte)(0x80 | ((code >> 6) & 0x3F)));
+ writeByte((byte)(0x80 | (code & 0x3F)));
+ }
+ }
+ }
+
/** Forces any buffered output to be written. */
public abstract void flush() throws IOException;
Index: src/demo/org/apache/lucene/demo/IndexLineFiles.java
===================================================================
--- src/demo/org/apache/lucene/demo/IndexLineFiles.java (revision 0)
+++ src/demo/org/apache/lucene/demo/IndexLineFiles.java (revision 0)
@@ -0,0 +1,271 @@
+package org.apache.lucene.demo;
+
+/**
+ * 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 org.apache.lucene.analysis.standard.StandardAnalyzer;
+import org.apache.lucene.analysis.SimpleSpaceAnalyzer;
+import org.apache.lucene.analysis.WhitespaceAnalyzer;
+import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.store.IndexOutput;
+import org.apache.lucene.store.FSDirectory;
+import org.apache.lucene.document.DateTools;
+
+import java.io.File;
+import java.io.Reader;
+import java.io.FileReader;
+import java.io.BufferedReader;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.Date;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+/** Index all text files under a directory. */
+public class IndexLineFiles {
+
+ private final static class ReusableStringReader extends Reader {
+ int upto;
+ int left;
+ char[] b = new char[128];
+ void init(char[] b, int offset, int len) {
+ if (len > this.b.length)
+ this.b = new char[(int) (len*1.25)];
+ System.arraycopy(b, offset, this.b, 0, len);
+ left = len;
+ this.upto = 0;
+ }
+ public int read(char[] c) {
+ return read(c, 0, c.length);
+ }
+ public int read(char[] c, int off, int len) {
+ if (left > len) {
+ System.arraycopy(b, upto, c, off, len);
+ upto += len;
+ left -= len;
+ return len;
+ } else if (0 == left) {
+ return -1;
+ } else {
+ System.arraycopy(b, upto, c, off, left);
+ upto += left;
+ int r = left;
+ left = 0;
+ return r;
+ }
+ }
+ public void close() {};
+ }
+
+ private IndexLineFiles() {}
+
+ static final File INDEX_DIR = new File("index");
+
+ static final AtomicInteger allDocCount = new AtomicInteger();
+
+ static int bufferSize;
+ static String fileName;
+
+ private static class Indexer extends Thread {
+
+ ReusableStringReader docReaders[] = new ReusableStringReader[mult];
+
+ Document doc = new Document();
+
+ int iter;
+
+ public void add(char[] b, int offset, int len) throws IOException {
+ //System.out.println("add: " + new String(b, offset, len));
+ docReaders[iter].init(b, offset, len);
+ if (++iter == mult) {
+ writer.addDocument(doc);
+ iter = 0;
+ allDocCount.getAndIncrement();
+ }
+ }
+
+ public void run() {
+
+ if (doStoredFields && 0 == iter) {
+ // Add the path of the file as a field named "path". Use a field that is
+ // indexed (i.e. searchable), but don't tokenize the field into words.
+ doc.add(new Field("path", fileName, Field.Store.YES, Field.Index.NO));
+
+ // Add the last modified date of the file a field named "modified". Use
+ // a field that is indexed (i.e. searchable), but don't tokenize the field
+ // into words.
+ doc.add(new Field("modified",
+ "200703161637",
+ Field.Store.YES, Field.Index.NO));
+ }
+
+ int iter = 0;
+ char[] buffer = new char[131072];
+
+ for(int i=0;i 0) {
+ add(buffer, 0, bufUpto);
+ if (allDocCount.get() >= numDoc) {
+ System.out.println("THREAD DONE");
+ return;
+ }
+ bufUpto = 0;
+ }
+ break;
+ }
+
+ // Break @ newlines:
+ final int len = bufUpto + numRead;
+ //System.out.println("read " + numRead + " now len=" + len);
+ int lineStart = 0;
+ for(int i=bufUpto;i