+++ /dev/null
-package org.apache.lucene.index;
-
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-import org.apache.lucene.index.FieldInfo.IndexOptions;
-import org.apache.lucene.util.UnicodeUtil;
-
-import java.io.IOException;
-import java.util.Collection;
-import java.util.Map;
-import java.util.ArrayList;
-import java.util.List;
-import org.apache.lucene.util.BitVector;
-import org.apache.lucene.util.CollectionUtil;
-
-final class FreqProxTermsWriter extends TermsHashConsumer {
-
- @Override
- public TermsHashConsumerPerThread addThread(TermsHashPerThread perThread) {
- return new FreqProxTermsWriterPerThread(perThread);
- }
-
- private static int compareText(final char[] text1, int pos1, final char[] text2, int pos2) {
- while(true) {
- final char c1 = text1[pos1++];
- final char c2 = text2[pos2++];
- if (c1 != c2) {
- if (0xffff == c2)
- return 1;
- else if (0xffff == c1)
- return -1;
- else
- return c1-c2;
- } else if (0xffff == c1)
- return 0;
- }
- }
-
- @Override
- void abort() {}
-
- // TODO: would be nice to factor out more of this, eg the
- // FreqProxFieldMergeState, and code to visit all Fields
- // under the same FieldInfo together, up into TermsHash*.
- // Other writers would presumably share alot of this...
-
- @Override
- public void flush(Map<TermsHashConsumerPerThread,Collection<TermsHashConsumerPerField>> threadsAndFields, final SegmentWriteState state) throws IOException {
-
- // Gather all FieldData's that have postings, across all
- // ThreadStates
- List<FreqProxTermsWriterPerField> allFields = new ArrayList<FreqProxTermsWriterPerField>();
-
- for (Map.Entry<TermsHashConsumerPerThread,Collection<TermsHashConsumerPerField>> entry : threadsAndFields.entrySet()) {
-
- Collection<TermsHashConsumerPerField> fields = entry.getValue();
-
- for (final TermsHashConsumerPerField i : fields) {
- final FreqProxTermsWriterPerField perField = (FreqProxTermsWriterPerField) i;
- if (perField.termsHashPerField.numPostings > 0)
- allFields.add(perField);
- }
- }
-
- // Sort by field name
- CollectionUtil.quickSort(allFields);
- final int numAllFields = allFields.size();
-
- // TODO: allow Lucene user to customize this consumer:
- final FormatPostingsFieldsConsumer consumer = new FormatPostingsFieldsWriter(state, fieldInfos);
- /*
- Current writer chain:
- FormatPostingsFieldsConsumer
- -> IMPL: FormatPostingsFieldsWriter
- -> FormatPostingsTermsConsumer
- -> IMPL: FormatPostingsTermsWriter
- -> FormatPostingsDocConsumer
- -> IMPL: FormatPostingsDocWriter
- -> FormatPostingsPositionsConsumer
- -> IMPL: FormatPostingsPositionsWriter
- */
- try {
- int start = 0;
- while(start < numAllFields) {
- final FieldInfo fieldInfo = allFields.get(start).fieldInfo;
- final String fieldName = fieldInfo.name;
-
- int end = start+1;
- while(end < numAllFields && allFields.get(end).fieldInfo.name.equals(fieldName))
- end++;
-
- FreqProxTermsWriterPerField[] fields = new FreqProxTermsWriterPerField[end-start];
- for(int i=start;i<end;i++) {
- fields[i-start] = allFields.get(i);
-
- // Aggregate the storePayload as seen by the same
- // field across multiple threads
- if (fieldInfo.indexOptions == IndexOptions.DOCS_AND_FREQS_AND_POSITIONS) {
- fieldInfo.storePayloads |= fields[i-start].hasPayloads;
- }
- }
-
- // If this field has postings then add them to the
- // segment
- appendPostings(fieldName, state, fields, consumer);
-
- for(int i=0;i<fields.length;i++) {
- TermsHashPerField perField = fields[i].termsHashPerField;
- int numPostings = perField.numPostings;
- perField.reset();
- perField.shrinkHash(numPostings);
- fields[i].reset();
- }
-
- start = end;
- }
-
- for (Map.Entry<TermsHashConsumerPerThread,Collection<TermsHashConsumerPerField>> entry : threadsAndFields.entrySet()) {
- FreqProxTermsWriterPerThread perThread = (FreqProxTermsWriterPerThread) entry.getKey();
- perThread.termsHashPerThread.reset(true);
- }
- } finally {
- consumer.finish();
- }
- }
-
- private byte[] payloadBuffer;
-
- /* Walk through all unique text tokens (Posting
- * instances) found in this field and serialize them
- * into a single RAM segment. */
- void appendPostings(String fieldName, SegmentWriteState state,
- FreqProxTermsWriterPerField[] fields,
- FormatPostingsFieldsConsumer consumer)
- throws CorruptIndexException, IOException {
-
- int numFields = fields.length;
-
- final FreqProxFieldMergeState[] mergeStates = new FreqProxFieldMergeState[numFields];
-
- for(int i=0;i<numFields;i++) {
- FreqProxFieldMergeState fms = mergeStates[i] = new FreqProxFieldMergeState(fields[i]);
-
- assert fms.field.fieldInfo == fields[0].fieldInfo;
-
- // Should always be true
- boolean result = fms.nextTerm();
- assert result;
- }
-
- final FormatPostingsTermsConsumer termsConsumer = consumer.addField(fields[0].fieldInfo);
- final Term protoTerm = new Term(fieldName);
-
- FreqProxFieldMergeState[] termStates = new FreqProxFieldMergeState[numFields];
-
- final IndexOptions currentFieldIndexOptions = fields[0].fieldInfo.indexOptions;
-
- final Map<Term,Integer> segDeletes;
- if (state.segDeletes != null && state.segDeletes.terms.size() > 0) {
- segDeletes = state.segDeletes.terms;
- } else {
- segDeletes = null;
- }
-
- try {
- // TODO: really TermsHashPerField should take over most
- // of this loop, including merge sort of terms from
- // multiple threads and interacting with the
- // TermsConsumer, only calling out to us (passing us the
- // DocsConsumer) to handle delivery of docs/positions
- while(numFields > 0) {
-
- // Get the next term to merge
- termStates[0] = mergeStates[0];
- int numToMerge = 1;
-
- // TODO: pqueue
- for(int i=1;i<numFields;i++) {
- final char[] text = mergeStates[i].text;
- final int textOffset = mergeStates[i].textOffset;
- final int cmp = compareText(text, textOffset, termStates[0].text, termStates[0].textOffset);
-
- if (cmp < 0) {
- termStates[0] = mergeStates[i];
- numToMerge = 1;
- } else if (cmp == 0)
- termStates[numToMerge++] = mergeStates[i];
- }
-
- final FormatPostingsDocsConsumer docConsumer = termsConsumer.addTerm(termStates[0].text, termStates[0].textOffset);
-
- final int delDocLimit;
- if (segDeletes != null) {
- final Integer docIDUpto = segDeletes.get(protoTerm.createTerm(termStates[0].termText()));
- if (docIDUpto != null) {
- delDocLimit = docIDUpto;
- } else {
- delDocLimit = 0;
- }
- } else {
- delDocLimit = 0;
- }
-
- try {
- // Now termStates has numToMerge FieldMergeStates
- // which all share the same term. Now we must
- // interleave the docID streams.
- while(numToMerge > 0) {
-
- FreqProxFieldMergeState minState = termStates[0];
- for(int i=1;i<numToMerge;i++)
- if (termStates[i].docID < minState.docID)
- minState = termStates[i];
-
- final int termDocFreq = minState.termFreq;
-
- final FormatPostingsPositionsConsumer posConsumer = docConsumer.addDoc(minState.docID, termDocFreq);
-
- // NOTE: we could check here if the docID was
- // deleted, and skip it. However, this is somewhat
- // dangerous because it can yield non-deterministic
- // behavior since we may see the docID before we see
- // the term that caused it to be deleted. This
- // would mean some (but not all) of its postings may
- // make it into the index, which'd alter the docFreq
- // for those terms. We could fix this by doing two
- // passes, ie first sweep marks all del docs, and
- // 2nd sweep does the real flush, but I suspect
- // that'd add too much time to flush.
-
- if (minState.docID < delDocLimit) {
- // Mark it deleted. TODO: we could also skip
- // writing its postings; this would be
- // deterministic (just for this Term's docs).
- if (state.deletedDocs == null) {
- state.deletedDocs = new BitVector(state.numDocs);
- }
- state.deletedDocs.set(minState.docID);
- }
-
- final ByteSliceReader prox = minState.prox;
-
- // Carefully copy over the prox + payload info,
- // changing the format to match Lucene's segment
- // format.
- if (currentFieldIndexOptions == IndexOptions.DOCS_AND_FREQS_AND_POSITIONS) {
- // omitTermFreqAndPositions == false so we do write positions &
- // payload
- try {
- int position = 0;
- for(int j=0;j<termDocFreq;j++) {
- final int code = prox.readVInt();
- position += code >> 1;
-
- final int payloadLength;
- if ((code & 1) != 0) {
- // This position has a payload
- payloadLength = prox.readVInt();
-
- if (payloadBuffer == null || payloadBuffer.length < payloadLength)
- payloadBuffer = new byte[payloadLength];
-
- prox.readBytes(payloadBuffer, 0, payloadLength);
-
- } else
- payloadLength = 0;
-
- posConsumer.addPosition(position, payloadBuffer, 0, payloadLength);
- } //End for
- } finally {
- posConsumer.finish();
- }
- }
-
- if (!minState.nextDoc()) {
-
- // Remove from termStates
- int upto = 0;
- for(int i=0;i<numToMerge;i++)
- if (termStates[i] != minState)
- termStates[upto++] = termStates[i];
- numToMerge--;
- assert upto == numToMerge;
-
- // Advance this state to the next term
-
- if (!minState.nextTerm()) {
- // OK, no more terms, so remove from mergeStates
- // as well
- upto = 0;
- for(int i=0;i<numFields;i++)
- if (mergeStates[i] != minState)
- mergeStates[upto++] = mergeStates[i];
- numFields--;
- assert upto == numFields;
- }
- }
- }
- } finally {
- docConsumer.finish();
- }
- }
- } finally {
- termsConsumer.finish();
- }
- }
-
- final UnicodeUtil.UTF8Result termsUTF8 = new UnicodeUtil.UTF8Result();
-}
fnp / pylucene.git / blobdiff