You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by sh...@apache.org on 2012/12/23 09:49:01 UTC
svn commit: r1425441 [2/5] - in /lucene/dev/branches/lucene4258/lucene:
codecs/src/java/org/apache/lucene/codecs/simpletext/
core/src/java/org/apache/lucene/codecs/
core/src/java/org/apache/lucene/codecs/compressing/
core/src/java/org/apache/lucene/cod...
Modified: lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/DocumentsWriter.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/DocumentsWriter.java?rev=1425441&r1=1425440&r2=1425441&view=diff
==============================================================================
--- lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/DocumentsWriter.java (original)
+++ lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/DocumentsWriter.java Sun Dec 23 08:49:00 2012
@@ -20,10 +20,13 @@ package org.apache.lucene.index;
import java.io.IOException;
import java.util.Collection;
import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.codecs.Codec;
+import org.apache.lucene.codecs.GenerationReplacementsFormat;
import org.apache.lucene.index.DocumentsWriterFlushQueue.SegmentFlushTicket;
import org.apache.lucene.index.DocumentsWriterPerThread.FlushedSegment;
import org.apache.lucene.index.DocumentsWriterPerThread.IndexingChain;
@@ -33,89 +36,75 @@ import org.apache.lucene.search.Query;
import org.apache.lucene.search.similarities.Similarity;
import org.apache.lucene.store.AlreadyClosedException;
import org.apache.lucene.store.Directory;
-import org.apache.lucene.store.FlushInfo;
import org.apache.lucene.store.IOContext;
import org.apache.lucene.util.InfoStream;
-import org.apache.lucene.util.MutableBits;
/**
- * This class accepts multiple added documents and directly
- * writes segment files.
- *
- * Each added document is passed to the {@link DocConsumer},
- * which in turn processes the document and interacts with
- * other consumers in the indexing chain. Certain
- * consumers, like {@link StoredFieldsConsumer} and {@link
- * TermVectorsConsumer}, digest a document and
- * immediately write bytes to the "doc store" files (ie,
- * they do not consume RAM per document, except while they
- * are processing the document).
- *
- * Other consumers, eg {@link FreqProxTermsWriter} and
- * {@link NormsConsumer}, buffer bytes in RAM and flush only
- * when a new segment is produced.
-
- * Once we have used our allowed RAM buffer, or the number
- * of added docs is large enough (in the case we are
- * flushing by doc count instead of RAM usage), we create a
- * real segment and flush it to the Directory.
- *
+ * This class accepts multiple added documents and directly writes segment
+ * files.
+ *
+ * Each added document is passed to the {@link DocConsumer}, which in turn
+ * processes the document and interacts with other consumers in the indexing
+ * chain. Certain consumers, like {@link StoredFieldsConsumer} and
+ * {@link TermVectorsConsumer}, digest a document and immediately write bytes to
+ * the "doc store" files (ie, they do not consume RAM per document, except while
+ * they are processing the document).
+ *
+ * Other consumers, eg {@link FreqProxTermsWriter} and {@link NormsConsumer},
+ * buffer bytes in RAM and flush only when a new segment is produced.
+ *
+ * Once we have used our allowed RAM buffer, or the number of added docs is
+ * large enough (in the case we are flushing by doc count instead of RAM usage),
+ * we create a real segment and flush it to the Directory.
+ *
* Threads:
- *
- * Multiple threads are allowed into addDocument at once.
- * There is an initial synchronized call to getThreadState
- * which allocates a ThreadState for this thread. The same
- * thread will get the same ThreadState over time (thread
- * affinity) so that if there are consistent patterns (for
- * example each thread is indexing a different content
- * source) then we make better use of RAM. Then
- * processDocument is called on that ThreadState without
- * synchronization (most of the "heavy lifting" is in this
- * call). Finally the synchronized "finishDocument" is
- * called to flush changes to the directory.
- *
- * When flush is called by IndexWriter we forcefully idle
- * all threads and flush only once they are all idle. This
- * means you can call flush with a given thread even while
- * other threads are actively adding/deleting documents.
- *
- *
+ *
+ * Multiple threads are allowed into addDocument at once. There is an initial
+ * synchronized call to getThreadState which allocates a ThreadState for this
+ * thread. The same thread will get the same ThreadState over time (thread
+ * affinity) so that if there are consistent patterns (for example each thread
+ * is indexing a different content source) then we make better use of RAM. Then
+ * processDocument is called on that ThreadState without synchronization (most
+ * of the "heavy lifting" is in this call). Finally the synchronized
+ * "finishDocument" is called to flush changes to the directory.
+ *
+ * When flush is called by IndexWriter we forcefully idle all threads and flush
+ * only once they are all idle. This means you can call flush with a given
+ * thread even while other threads are actively adding/deleting documents.
+ *
+ *
* Exceptions:
- *
- * Because this class directly updates in-memory posting
- * lists, and flushes stored fields and term vectors
- * directly to files in the directory, there are certain
- * limited times when an exception can corrupt this state.
- * For example, a disk full while flushing stored fields
- * leaves this file in a corrupt state. Or, an OOM
- * exception while appending to the in-memory posting lists
- * can corrupt that posting list. We call such exceptions
- * "aborting exceptions". In these cases we must call
- * abort() to discard all docs added since the last flush.
- *
- * All other exceptions ("non-aborting exceptions") can
- * still partially update the index structures. These
- * updates are consistent, but, they represent only a part
- * of the document seen up until the exception was hit.
- * When this happens, we immediately mark the document as
- * deleted so that the document is always atomically ("all
- * or none") added to the index.
+ *
+ * Because this class directly updates in-memory posting lists, and flushes
+ * stored fields and term vectors directly to files in the directory, there are
+ * certain limited times when an exception can corrupt this state. For example,
+ * a disk full while flushing stored fields leaves this file in a corrupt state.
+ * Or, an OOM exception while appending to the in-memory posting lists can
+ * corrupt that posting list. We call such exceptions "aborting exceptions". In
+ * these cases we must call abort() to discard all docs added since the last
+ * flush.
+ *
+ * All other exceptions ("non-aborting exceptions") can still partially update
+ * the index structures. These updates are consistent, but, they represent only
+ * a part of the document seen up until the exception was hit. When this
+ * happens, we immediately mark the document as deleted so that the document is
+ * always atomically ("all or none") added to the index.
*/
final class DocumentsWriter {
Directory directory;
-
+
private volatile boolean closed;
-
+
final InfoStream infoStream;
Similarity similarity;
-
+
List<String> newFiles;
-
+
final IndexWriter indexWriter;
-
+
private AtomicInteger numDocsInRAM = new AtomicInteger(0);
-
+
// TODO: cut over to BytesRefHash in BufferedDeletes
volatile DocumentsWriterDeleteQueue deleteQueue = new DocumentsWriterDeleteQueue();
private final DocumentsWriterFlushQueue ticketQueue = new DocumentsWriterFlushQueue();
@@ -126,17 +115,20 @@ final class DocumentsWriter {
* #anyChanges() & #flushAllThreads
*/
private volatile boolean pendingChangesInCurrentFullFlush;
-
- private Collection<String> abortedFiles; // List of files that were written before last abort()
-
+
+ private Collection<String> abortedFiles; // List of files that were written
+ // before last abort()
+
final IndexingChain chain;
-
+
final DocumentsWriterPerThreadPool perThreadPool;
final FlushPolicy flushPolicy;
final DocumentsWriterFlushControl flushControl;
final Codec codec;
- DocumentsWriter(Codec codec, LiveIndexWriterConfig config, Directory directory, IndexWriter writer, FieldNumbers globalFieldNumbers,
+
+ DocumentsWriter(Codec codec, LiveIndexWriterConfig config,
+ Directory directory, IndexWriter writer, FieldNumbers globalFieldNumbers,
BufferedDeletesStream bufferedDeletesStream) {
this.codec = codec;
this.directory = directory;
@@ -151,7 +143,7 @@ final class DocumentsWriter {
flushPolicy.init(this);
flushControl = new DocumentsWriterFlushControl(this, config);
}
-
+
synchronized void deleteQueries(final Query... queries) throws IOException {
deleteQueue.addDelete(queries);
flushControl.doOnDelete();
@@ -159,7 +151,7 @@ final class DocumentsWriter {
applyAllDeletes(deleteQueue);
}
}
-
+
// TODO: we could check w/ FreqProxTermsWriter: if the
// term doesn't exist, don't bother buffering into the
// per-DWPT map (but still must go into the global map)
@@ -171,47 +163,49 @@ final class DocumentsWriter {
applyAllDeletes(deleteQueue);
}
}
-
+
DocumentsWriterDeleteQueue currentDeleteSession() {
return deleteQueue;
}
- private void applyAllDeletes(DocumentsWriterDeleteQueue deleteQueue) throws IOException {
+ private void applyAllDeletes(DocumentsWriterDeleteQueue deleteQueue)
+ throws IOException {
if (deleteQueue != null && !flushControl.isFullFlush()) {
ticketQueue.addDeletesAndPurge(this, deleteQueue);
}
indexWriter.applyAllDeletes();
indexWriter.flushCount.incrementAndGet();
}
-
+
/** Returns how many docs are currently buffered in RAM. */
int getNumDocs() {
return numDocsInRAM.get();
}
-
+
Collection<String> abortedFiles() {
return abortedFiles;
}
-
+
private void ensureOpen() throws AlreadyClosedException {
if (closed) {
throw new AlreadyClosedException("this IndexWriter is closed");
}
}
-
- /** Called if we hit an exception at a bad time (when
- * updating the index files) and must discard all
- * currently buffered docs. This resets our state,
- * discarding any docs added since last flush. */
+
+ /**
+ * Called if we hit an exception at a bad time (when updating the index files)
+ * and must discard all currently buffered docs. This resets our state,
+ * discarding any docs added since last flush.
+ */
synchronized void abort() {
boolean success = false;
-
+
try {
deleteQueue.clear();
if (infoStream.isEnabled("DW")) {
infoStream.message("DW", "abort");
}
-
+
final int limit = perThreadPool.getActiveThreadState();
for (int i = 0; i < limit; i++) {
final ThreadState perThread = perThreadPool.getThreadState(i);
@@ -236,53 +230,58 @@ final class DocumentsWriter {
success = true;
} finally {
if (infoStream.isEnabled("DW")) {
- infoStream.message("DW", "done abort; abortedFiles=" + abortedFiles + " success=" + success);
+ infoStream.message("DW", "done abort; abortedFiles=" + abortedFiles
+ + " success=" + success);
}
}
}
-
+
boolean anyChanges() {
if (infoStream.isEnabled("DW")) {
- infoStream.message("DW", "anyChanges? numDocsInRam=" + numDocsInRAM.get()
- + " deletes=" + anyDeletions() + " hasTickets:"
- + ticketQueue.hasTickets() + " pendingChangesInFullFlush: "
- + pendingChangesInCurrentFullFlush);
+ infoStream.message("DW",
+ "anyChanges? numDocsInRam=" + numDocsInRAM.get() + " deletes="
+ + anyDeletions() + " hasTickets:" + ticketQueue.hasTickets()
+ + " pendingChangesInFullFlush: "
+ + pendingChangesInCurrentFullFlush);
}
/*
- * changes are either in a DWPT or in the deleteQueue.
- * yet if we currently flush deletes and / or dwpt there
- * could be a window where all changes are in the ticket queue
- * before they are published to the IW. ie we need to check if the
- * ticket queue has any tickets.
+ * changes are either in a DWPT or in the deleteQueue. yet if we currently
+ * flush deletes and / or dwpt there could be a window where all changes are
+ * in the ticket queue before they are published to the IW. ie we need to
+ * check if the ticket queue has any tickets.
*/
- return numDocsInRAM.get() != 0 || anyDeletions() || ticketQueue.hasTickets() || pendingChangesInCurrentFullFlush;
+ return numDocsInRAM.get() != 0 || anyDeletions()
+ || ticketQueue.hasTickets() || pendingChangesInCurrentFullFlush;
}
public int getBufferedDeleteTermsSize() {
return deleteQueue.getBufferedDeleteTermsSize();
}
-
- //for testing
+
+ // for testing
public int getNumBufferedDeleteTerms() {
return deleteQueue.numGlobalTermDeletes();
}
-
+
public boolean anyDeletions() {
return deleteQueue.anyChanges();
}
-
+
void close() {
closed = true;
flushControl.setClosed();
}
-
+
private boolean preUpdate() throws IOException {
ensureOpen();
boolean maybeMerge = false;
if (flushControl.anyStalledThreads() || flushControl.numQueuedFlushes() > 0) {
// Help out flushing any queued DWPTs so we can un-stall:
if (infoStream.isEnabled("DW")) {
- infoStream.message("DW", "DocumentsWriter has queued dwpt; will hijack this thread to flush pending segment(s)");
+ infoStream
+ .message(
+ "DW",
+ "DocumentsWriter has queued dwpt; will hijack this thread to flush pending segment(s)");
}
do {
// Try pick up pending threads here if possible
@@ -291,52 +290,58 @@ final class DocumentsWriter {
// Don't push the delete here since the update could fail!
maybeMerge |= doFlush(flushingDWPT);
}
-
+
if (infoStream.isEnabled("DW")) {
if (flushControl.anyStalledThreads()) {
- infoStream.message("DW", "WARNING DocumentsWriter has stalled threads; waiting");
+ infoStream.message("DW",
+ "WARNING DocumentsWriter has stalled threads; waiting");
}
}
flushControl.waitIfStalled(); // block if stalled
- } while (flushControl.numQueuedFlushes() != 0); // still queued DWPTs try help flushing
-
+ } while (flushControl.numQueuedFlushes() != 0); // still queued DWPTs try
+ // help flushing
+
if (infoStream.isEnabled("DW")) {
- infoStream.message("DW", "continue indexing after helping out flushing DocumentsWriter is healthy");
+ infoStream
+ .message("DW",
+ "continue indexing after helping out flushing DocumentsWriter is healthy");
}
}
return maybeMerge;
}
-
- private boolean postUpdate(DocumentsWriterPerThread flushingDWPT, boolean maybeMerge) throws IOException {
+
+ private boolean postUpdate(DocumentsWriterPerThread flushingDWPT,
+ boolean maybeMerge) throws IOException {
if (flushControl.doApplyAllDeletes()) {
applyAllDeletes(deleteQueue);
}
if (flushingDWPT != null) {
maybeMerge |= doFlush(flushingDWPT);
} else {
- final DocumentsWriterPerThread nextPendingFlush = flushControl.nextPendingFlush();
+ final DocumentsWriterPerThread nextPendingFlush = flushControl
+ .nextPendingFlush();
if (nextPendingFlush != null) {
maybeMerge |= doFlush(nextPendingFlush);
}
}
-
+
return maybeMerge;
}
-
- boolean updateDocuments(final Iterable<? extends IndexDocument> docs, final Analyzer analyzer,
- final Term delTerm) throws IOException {
+
+ boolean updateDocuments(final Iterable<? extends IndexDocument> docs,
+ final Analyzer analyzer, final Term delTerm) throws IOException {
boolean maybeMerge = preUpdate();
-
+
final ThreadState perThread = flushControl.obtainAndLock();
final DocumentsWriterPerThread flushingDWPT;
try {
if (!perThread.isActive()) {
ensureOpen();
- assert false: "perThread is not active but we are still open";
+ assert false : "perThread is not active but we are still open";
}
-
+
final DocumentsWriterPerThread dwpt = perThread.dwpt;
try {
final int docCount = dwpt.updateDocuments(docs, analyzer, delTerm);
@@ -351,29 +356,30 @@ final class DocumentsWriter {
} finally {
perThread.unlock();
}
-
+
return postUpdate(flushingDWPT, maybeMerge);
}
-
+
boolean updateDocument(final IndexDocument doc, final Analyzer analyzer,
final Term delTerm) throws IOException {
-
+
boolean maybeMerge = preUpdate();
-
+
final ThreadState perThread = flushControl.obtainAndLock();
-
+
final DocumentsWriterPerThread flushingDWPT;
try {
-
+
if (!perThread.isActive()) {
ensureOpen();
- throw new IllegalStateException("perThread is not active but we are still open");
+ throw new IllegalStateException(
+ "perThread is not active but we are still open");
}
-
+
final DocumentsWriterPerThread dwpt = perThread.dwpt;
try {
- dwpt.updateDocument(doc, analyzer, delTerm);
+ dwpt.updateDocument(doc, analyzer, delTerm);
numDocsInRAM.incrementAndGet();
} finally {
if (dwpt.checkAndResetHasAborted()) {
@@ -385,20 +391,104 @@ final class DocumentsWriter {
} finally {
perThread.unlock();
}
-
+
return postUpdate(flushingDWPT, maybeMerge);
}
-
- private boolean doFlush(DocumentsWriterPerThread flushingDWPT) throws IOException {
- boolean maybeMerge = false;
+
+ boolean updateFields(Term term, FieldsUpdate fieldsUpdate) throws IOException {
+ boolean maybeMerge = preUpdate();
+
+ final ThreadState perThread = flushControl.obtainAndLock();
+
+ final DocumentsWriterPerThread flushingDWPT;
+
+ try {
+
+ if (!perThread.isActive()) {
+ ensureOpen();
+ throw new IllegalStateException(
+ "perThread is not active but we are still open");
+ }
+
+ final DocumentsWriterPerThread dwpt = perThread.dwpt;
+ try {
+ dwpt.updateFields(term, fieldsUpdate);
+ } finally {
+ if (dwpt.checkAndResetHasAborted()) {
+ flushControl.doOnAbort(perThread);
+ }
+ }
+ final boolean isUpdate = term != null;
+ flushingDWPT = flushControl.doAfterDocument(perThread, isUpdate);
+ } finally {
+ perThread.unlock();
+ }
+ return postUpdate(flushingDWPT, maybeMerge);
+ }
+
+ void writeUpdatedSegment(UpdatedSegmentData liveUpdates,
+ SegmentInfoPerCommit info, IndexFileDeleter deleter) throws IOException {
+ final ThreadState perThread = flushControl.obtainAndLock();
+
+ try {
+ if (!perThread.isActive()) {
+ ensureOpen();
+ throw new IllegalStateException(
+ "perThread is not active but we are still open");
+ }
+
+ final DocumentsWriterPerThread dwpt = perThread.dwpt;
+ try {
+ // start new segment, with update generation in name
+ dwpt.initSegmentInfo(info.info, info.getNextUpdateGen());
+
+ // push documents, including empty ones where needed
+ liveUpdates.startWriting(info.getNextUpdateGen(),
+ info.info.getDocCount());
+ IndexDocument doc;
+ while ((doc = liveUpdates.nextDocument()) != null) {
+ dwpt.updateDocument(doc, liveUpdates.getAnalyzer(), null);
+ }
+
+ // add field generation replacements
+ final Map<String,FieldGenerationReplacements> generationReplacments = liveUpdates
+ .getFieldGenerationReplacments();
+ if (generationReplacments != null) {
+ for (Entry<String,FieldGenerationReplacements> field : generationReplacments
+ .entrySet()) {
+ final GenerationReplacementsFormat repsFormat = codec
+ .generationReplacementsFormat();
+ repsFormat.writeGenerationReplacement(field.getKey(),
+ field.getValue(), directory, info, IOContext.DEFAULT);
+ }
+ }
+
+ // flush directly
+ dwpt.clearDeleteSlice();
+ dwpt.flush(info.getNextUpdateGen());
+ } finally {
+ if (dwpt.checkAndResetHasAborted()) {
+ flushControl.doOnAbort(perThread);
+ }
+ }
+ } finally {
+ perThread.unlock();
+ }
+ }
+
+ private boolean doFlush(DocumentsWriterPerThread flushingDWPT)
+ throws IOException {
+ int actualFlushes = 0;
while (flushingDWPT != null) {
- maybeMerge = true;
+ actualFlushes++;
boolean success = false;
SegmentFlushTicket ticket = null;
try {
assert currentFullFlushDelQueue == null
|| flushingDWPT.deleteQueue == currentFullFlushDelQueue : "expected: "
- + currentFullFlushDelQueue + "but was: " + flushingDWPT.deleteQueue
+ + currentFullFlushDelQueue
+ + "but was: "
+ + flushingDWPT.deleteQueue
+ " " + flushControl.isFullFlush();
/*
* Since with DWPT the flush process is concurrent and several DWPT
@@ -415,18 +505,25 @@ final class DocumentsWriter {
* might miss to deletes documents in 'A'.
*/
try {
- // Each flush is assigned a ticket in the order they acquire the ticketQueue lock
+ // Each flush is assigned a ticket in the order they acquire the
+ // ticketQueue lock
ticket = ticketQueue.addFlushTicket(flushingDWPT);
-
+
// flush concurrently without locking
- final FlushedSegment newSegment = flushingDWPT.flush();
- ticketQueue.addSegment(ticket, newSegment);
- // flush was successful once we reached this point - new seg. has been assigned to the ticket!
- success = true;
+ final FlushedSegment newSegment = flushingDWPT.flush(-1);
+ if (newSegment == null) {
+ actualFlushes--;
+ } else {
+ ticketQueue.addSegment(ticket, newSegment);
+ // flush was successful once we reached this point - new seg. has
+ // been assigned to the ticket!
+ success = true;
+ }
} finally {
if (!success && ticket != null) {
// In the case of a failure make sure we are making progress and
- // apply all the deletes since the segment flush failed since the flush
+ // apply all the deletes since the segment flush failed since the
+ // flush
// ticket could hold global deletes see FlushTicket#canPublish()
ticketQueue.markTicketFailed(ticket);
}
@@ -435,59 +532,63 @@ final class DocumentsWriter {
* Now we are done and try to flush the ticket queue if the head of the
* queue has already finished the flush.
*/
- if (ticketQueue.getTicketCount() >= perThreadPool.getActiveThreadState()) {
+ if (ticketQueue.getTicketCount() >= perThreadPool
+ .getActiveThreadState()) {
// This means there is a backlog: the one
// thread in innerPurge can't keep up with all
- // other threads flushing segments. In this case
+ // other threads flushing segments. In this case
// we forcefully stall the producers.
ticketQueue.forcePurge(this);
} else {
ticketQueue.tryPurge(this);
}
-
+
} finally {
flushControl.doAfterFlush(flushingDWPT);
flushingDWPT.checkAndResetHasAborted();
indexWriter.flushCount.incrementAndGet();
indexWriter.doAfterFlush();
}
-
+
flushingDWPT = flushControl.nextPendingFlush();
}
-
+
// If deletes alone are consuming > 1/2 our RAM
// buffer, force them all to apply now. This is to
// prevent too-frequent flushing of a long tail of
// tiny segments:
final double ramBufferSizeMB = indexWriter.getConfig().getRAMBufferSizeMB();
- if (ramBufferSizeMB != IndexWriterConfig.DISABLE_AUTO_FLUSH &&
- flushControl.getDeleteBytesUsed() > (1024*1024*ramBufferSizeMB/2)) {
+ if (ramBufferSizeMB != IndexWriterConfig.DISABLE_AUTO_FLUSH
+ && flushControl.getDeleteBytesUsed() > (1024 * 1024 * ramBufferSizeMB / 2)) {
if (infoStream.isEnabled("DW")) {
- infoStream.message("DW", "force apply deletes bytesUsed=" + flushControl.getDeleteBytesUsed() + " vs ramBuffer=" + (1024*1024*ramBufferSizeMB));
+ infoStream.message("DW", "force apply deletes bytesUsed="
+ + flushControl.getDeleteBytesUsed() + " vs ramBuffer="
+ + (1024 * 1024 * ramBufferSizeMB));
}
applyAllDeletes(deleteQueue);
}
-
- return maybeMerge;
+
+ return actualFlushes > 0;
}
-
- void finishFlush(FlushedSegment newSegment, FrozenBufferedDeletes bufferedDeletes)
- throws IOException {
+ void finishFlush(FlushedSegment newSegment,
+ FrozenBufferedDeletes bufferedDeletes) throws IOException {
// Finish the flushed segment and publish it to IndexWriter
if (newSegment == null) {
assert bufferedDeletes != null;
- if (bufferedDeletes != null && bufferedDeletes.any()) {
+ if (bufferedDeletes != null
+ && (bufferedDeletes.anyDeletes() || bufferedDeletes.anyUpdates())) {
indexWriter.publishFrozenDeletes(bufferedDeletes);
if (infoStream.isEnabled("DW")) {
- infoStream.message("DW", "flush: push buffered deletes: " + bufferedDeletes);
+ infoStream.message("DW", "flush: push buffered deletes: "
+ + bufferedDeletes);
}
}
} else {
- publishFlushedSegment(newSegment, bufferedDeletes);
+ publishFlushedSegment(newSegment, bufferedDeletes);
}
}
-
+
final void subtractFlushedNumDocs(int numFlushed) {
int oldValue = numDocsInRAM.get();
while (!numDocsInRAM.compareAndSet(oldValue, oldValue - numFlushed)) {
@@ -497,55 +598,62 @@ final class DocumentsWriter {
/**
* Publishes the flushed segment, segment private deletes (if any) and its
- * associated global delete (if present) to IndexWriter. The actual
- * publishing operation is synced on IW -> BDS so that the {@link SegmentInfo}'s
- * delete generation is always GlobalPacket_deleteGeneration + 1
+ * associated global delete (if present) to IndexWriter. The actual publishing
+ * operation is synced on IW -> BDS so that the {@link SegmentInfo}'s delete
+ * generation is always GlobalPacket_deleteGeneration + 1
*/
- private void publishFlushedSegment(FlushedSegment newSegment, FrozenBufferedDeletes globalPacket)
- throws IOException {
+ private void publishFlushedSegment(FlushedSegment newSegment,
+ FrozenBufferedDeletes globalPacket) throws IOException {
assert newSegment != null;
assert newSegment.segmentInfo != null;
final FrozenBufferedDeletes segmentDeletes = newSegment.segmentDeletes;
- //System.out.println("FLUSH: " + newSegment.segmentInfo.info.name);
+ // System.out.println("FLUSH: " + newSegment.segmentInfo.info.name);
if (infoStream.isEnabled("DW")) {
- infoStream.message("DW", "publishFlushedSegment seg-private deletes=" + segmentDeletes);
+ infoStream.message("DW", "publishFlushedSegment seg-private deletes="
+ + segmentDeletes);
}
-
if (segmentDeletes != null && infoStream.isEnabled("DW")) {
- infoStream.message("DW", "flush: push buffered seg private deletes: " + segmentDeletes);
+ infoStream.message("DW", "flush: push buffered seg private deletes: "
+ + segmentDeletes);
}
// now publish!
- indexWriter.publishFlushedSegment(newSegment.segmentInfo, segmentDeletes, globalPacket);
+ indexWriter.publishFlushedSegment(newSegment.segmentInfo, segmentDeletes,
+ globalPacket);
}
// for asserts
private volatile DocumentsWriterDeleteQueue currentFullFlushDelQueue = null;
-
+
// for asserts
- private synchronized boolean setFlushingDeleteQueue(DocumentsWriterDeleteQueue session) {
+ private synchronized boolean setFlushingDeleteQueue(
+ DocumentsWriterDeleteQueue session) {
currentFullFlushDelQueue = session;
return true;
}
/*
* FlushAllThreads is synced by IW fullFlushLock. Flushing all threads is a
- * two stage operation; the caller must ensure (in try/finally) that finishFlush
- * is called after this method, to release the flush lock in DWFlushControl
+ * two stage operation; the caller must ensure (in try/finally) that
+ * finishFlush is called after this method, to release the flush lock in
+ * DWFlushControl
*/
- final boolean flushAllThreads()
- throws IOException {
+ final boolean flushAllThreads() throws IOException {
final DocumentsWriterDeleteQueue flushingDeleteQueue;
if (infoStream.isEnabled("DW")) {
- infoStream.message("DW", Thread.currentThread().getName() + " startFullFlush");
+ infoStream.message("DW", Thread.currentThread().getName()
+ + " startFullFlush");
}
synchronized (this) {
pendingChangesInCurrentFullFlush = anyChanges();
flushingDeleteQueue = deleteQueue;
- /* Cutover to a new delete queue. This must be synced on the flush control
+ /*
+ * Cutover to a new delete queue. This must be synced on the flush control
* otherwise a new DWPT could sneak into the loop with an already flushing
- * delete queue */
- flushControl.markForFullFlush(); // swaps the delQueue synced on FlushControl
+ * delete queue
+ */
+ flushControl.markForFullFlush(); // swaps the delQueue synced on
+ // FlushControl
assert setFlushingDeleteQueue(flushingDeleteQueue);
}
assert currentFullFlushDelQueue != null;
@@ -559,10 +667,15 @@ final class DocumentsWriter {
anythingFlushed |= doFlush(flushingDWPT);
}
// If a concurrent flush is still in flight wait for it
- flushControl.waitForFlush();
- if (!anythingFlushed && flushingDeleteQueue.anyChanges()) { // apply deletes if we did not flush any document
+ flushControl.waitForFlush();
+ if (!anythingFlushed && flushingDeleteQueue.anyChanges()) { // apply
+ // deletes if
+ // we did not
+ // flush any
+ // document
if (infoStream.isEnabled("DW")) {
- infoStream.message("DW", Thread.currentThread().getName() + ": flush naked frozen global deletes");
+ infoStream.message("DW", Thread.currentThread().getName()
+ + ": flush naked frozen global deletes");
}
ticketQueue.addDeletesAndPurge(this, flushingDeleteQueue);
} else {
@@ -578,7 +691,8 @@ final class DocumentsWriter {
final void finishFullFlush(boolean success) {
try {
if (infoStream.isEnabled("DW")) {
- infoStream.message("DW", Thread.currentThread().getName() + " finishFullFlush success=" + success);
+ infoStream.message("DW", Thread.currentThread().getName()
+ + " finishFullFlush success=" + success);
}
assert setFlushingDeleteQueue(null);
if (success) {
Modified: lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/DocumentsWriterDeleteQueue.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/DocumentsWriterDeleteQueue.java?rev=1425441&r1=1425440&r2=1425441&view=diff
==============================================================================
--- lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/DocumentsWriterDeleteQueue.java (original)
+++ lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/DocumentsWriterDeleteQueue.java Sun Dec 23 08:49:00 2012
@@ -25,15 +25,15 @@ import org.apache.lucene.search.Query;
/**
* {@link DocumentsWriterDeleteQueue} is a non-blocking linked pending deletes
- * queue. In contrast to other queue implementation we only maintain the
- * tail of the queue. A delete queue is always used in a context of a set of
- * DWPTs and a global delete pool. Each of the DWPT and the global pool need to
- * maintain their 'own' head of the queue (as a DeleteSlice instance per DWPT).
- * The difference between the DWPT and the global pool is that the DWPT starts
- * maintaining a head once it has added its first document since for its segments
- * private deletes only the deletes after that document are relevant. The global
- * pool instead starts maintaining the head once this instance is created by
- * taking the sentinel instance as its initial head.
+ * queue. In contrast to other queue implementation we only maintain the tail of
+ * the queue. A delete queue is always used in a context of a set of DWPTs and a
+ * global delete pool. Each of the DWPT and the global pool need to maintain
+ * their 'own' head of the queue (as a DeleteSlice instance per DWPT). The
+ * difference between the DWPT and the global pool is that the DWPT starts
+ * maintaining a head once it has added its first document since for its
+ * segments private deletes only the deletes after that document are relevant.
+ * The global pool instead starts maintaining the head once this instance is
+ * created by taking the sentinel instance as its initial head.
* <p>
* Since each {@link DeleteSlice} maintains its own head and the list is only
* single linked the garbage collector takes care of pruning the list for us.
@@ -44,14 +44,15 @@ import org.apache.lucene.search.Query;
* Each DWPT as well as the global delete pool maintain their private
* DeleteSlice instance. In the DWPT case updating a slice is equivalent to
* atomically finishing the document. The slice update guarantees a "happens
- * before" relationship to all other updates in the same indexing session. When a
- * DWPT updates a document it:
+ * before" relationship to all other updates in the same indexing session. When
+ * a DWPT updates a document it:
*
* <ol>
* <li>consumes a document and finishes its processing</li>
* <li>updates its private {@link DeleteSlice} either by calling
- * {@link #updateSlice(DeleteSlice)} or {@link #add(Term, DeleteSlice)} (if the
- * document has a delTerm)</li>
+ * {@link #updateSlice(DeleteSlice)} or
+ * {@link #add(Term, DeleteSlice, FieldsUpdate)} (if the document has a delTerm)
+ * </li>
* <li>applies all deletes in the slice to its private {@link BufferedDeletes}
* and resets it</li>
* <li>increments its internal document id</li>
@@ -73,6 +74,7 @@ final class DocumentsWriterDeleteQueue {
private final DeleteSlice globalSlice;
private final BufferedDeletes globalBufferedDeletes;
+ private final BufferedUpdates globalBufferedUpdates;
/* only acquired to update the global deletes */
private final ReentrantLock globalBufferLock = new ReentrantLock();
@@ -83,11 +85,12 @@ final class DocumentsWriterDeleteQueue {
}
DocumentsWriterDeleteQueue(long generation) {
- this(new BufferedDeletes(), generation);
+ this(new BufferedDeletes(), new BufferedUpdates(), generation);
}
- DocumentsWriterDeleteQueue(BufferedDeletes globalBufferedDeletes, long generation) {
+ DocumentsWriterDeleteQueue(BufferedDeletes globalBufferedDeletes, BufferedUpdates globalBufferedUpdates, long generation) {
this.globalBufferedDeletes = globalBufferedDeletes;
+ this.globalBufferedUpdates = globalBufferedUpdates;
this.generation = generation;
/*
* we use a sentinel instance as our initial tail. No slice will ever try to
@@ -110,8 +113,8 @@ final class DocumentsWriterDeleteQueue {
/**
* invariant for document update
*/
- void add(Term term, DeleteSlice slice) {
- final TermNode termNode = new TermNode(term);
+ void add(Term term, DeleteSlice slice, FieldsUpdate fieldsUpdate) {
+ final TermNode termNode = new TermNode(term, fieldsUpdate);
// System.out.println(Thread.currentThread().getName() + ": push " + termNode + " this=" + this);
add(termNode);
/*
@@ -174,7 +177,7 @@ final class DocumentsWriterDeleteQueue {
* and if the global slice is up-to-date
* and if globalBufferedDeletes has changes
*/
- return globalBufferedDeletes.any() || !globalSlice.isEmpty() || globalSlice.sliceTail != tail
+ return globalBufferedDeletes.any() || globalBufferedUpdates.any() || !globalSlice.isEmpty() || globalSlice.sliceTail != tail
|| tail.next != null;
} finally {
globalBufferLock.unlock();
@@ -192,7 +195,7 @@ final class DocumentsWriterDeleteQueue {
try {
if (updateSlice(globalSlice)) {
// System.out.println(Thread.currentThread() + ": apply globalSlice");
- globalSlice.apply(globalBufferedDeletes, BufferedDeletes.MAX_INT);
+ globalSlice.apply(globalBufferedDeletes, globalBufferedUpdates, BufferedDeletes.MAX_INT);
}
} finally {
globalBufferLock.unlock();
@@ -217,13 +220,14 @@ final class DocumentsWriterDeleteQueue {
try {
if (globalSlice.sliceTail != currentTail) {
globalSlice.sliceTail = currentTail;
- globalSlice.apply(globalBufferedDeletes, BufferedDeletes.MAX_INT);
+ globalSlice.apply(globalBufferedDeletes, globalBufferedUpdates, BufferedDeletes.MAX_INT);
}
// System.out.println(Thread.currentThread().getName() + ": now freeze global buffer " + globalBufferedDeletes);
final FrozenBufferedDeletes packet = new FrozenBufferedDeletes(
- globalBufferedDeletes, false);
+ globalBufferedDeletes, globalBufferedUpdates, false);
globalBufferedDeletes.clear();
+ globalBufferedUpdates.clear();
return packet;
} finally {
globalBufferLock.unlock();
@@ -257,7 +261,7 @@ final class DocumentsWriterDeleteQueue {
sliceHead = sliceTail = currentTail;
}
- void apply(BufferedDeletes del, int docIDUpto) {
+ void apply(BufferedDeletes del, BufferedUpdates update, int docIDUpto) {
if (sliceHead == sliceTail) {
// 0 length slice
return;
@@ -272,7 +276,7 @@ final class DocumentsWriterDeleteQueue {
do {
current = current.next;
assert current != null : "slice property violated between the head on the tail must not be a null node";
- current.apply(del, docIDUpto);
+ current.apply(del, update, docIDUpto);
// System.out.println(Thread.currentThread().getName() + ": pull " + current + " docIDUpto=" + docIDUpto);
} while (current != sliceTail);
reset();
@@ -323,7 +327,7 @@ final class DocumentsWriterDeleteQueue {
static final AtomicReferenceFieldUpdater<Node,Node> nextUpdater = AtomicReferenceFieldUpdater
.newUpdater(Node.class, Node.class, "next");
- void apply(BufferedDeletes bufferedDeletes, int docIDUpto) {
+ void apply(BufferedDeletes bufferedDeletes, BufferedUpdates bufferedUpdates, int docIDUpto) {
throw new IllegalStateException("sentinel item must never be applied");
}
@@ -333,14 +337,20 @@ final class DocumentsWriterDeleteQueue {
}
private static final class TermNode extends Node<Term> {
-
- TermNode(Term term) {
+ FieldsUpdate fieldsUpdate;
+
+ TermNode(Term term, FieldsUpdate fieldsUpdate) {
super(term);
+ this.fieldsUpdate = fieldsUpdate;
}
@Override
- void apply(BufferedDeletes bufferedDeletes, int docIDUpto) {
- bufferedDeletes.addTerm(item, docIDUpto);
+ void apply(BufferedDeletes bufferedDeletes, BufferedUpdates bufferedUpdates, int docIDUpto) {
+ if (fieldsUpdate == null) {
+ bufferedDeletes.addTerm(item, docIDUpto);
+ } else {
+ bufferedUpdates.addTerm(item, new FieldsUpdate(fieldsUpdate, docIDUpto));
+ }
}
@Override
@@ -355,7 +365,7 @@ final class DocumentsWriterDeleteQueue {
}
@Override
- void apply(BufferedDeletes bufferedDeletes, int docIDUpto) {
+ void apply(BufferedDeletes bufferedDeletes, BufferedUpdates bufferedUpdates, int docIDUpto) {
for (Query query : item) {
bufferedDeletes.addQuery(query, docIDUpto);
}
@@ -368,7 +378,7 @@ final class DocumentsWriterDeleteQueue {
}
@Override
- void apply(BufferedDeletes bufferedDeletes, int docIDUpto) {
+ void apply(BufferedDeletes bufferedDeletes, BufferedUpdates bufferedUpdates, int docIDUpto) {
for (Term term : item) {
bufferedDeletes.addTerm(term, docIDUpto);
}
@@ -387,7 +397,7 @@ final class DocumentsWriterDeleteQueue {
try {
if (globalSlice.sliceTail != currentTail) {
globalSlice.sliceTail = currentTail;
- globalSlice.apply(globalBufferedDeletes, BufferedDeletes.MAX_INT);
+ globalSlice.apply(globalBufferedDeletes, globalBufferedUpdates, BufferedDeletes.MAX_INT);
}
return globalBufferedDeletes.any();
} finally {
@@ -409,6 +419,10 @@ final class DocumentsWriterDeleteQueue {
return globalBufferedDeletes.bytesUsed.get();
}
+ public long updateBytesUsed() {
+ return globalBufferedUpdates.bytesUsed.get();
+ }
+
@Override
public String toString() {
return "DWDQ: [ generation: " + generation + " ]";
Modified: lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/DocumentsWriterPerThread.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/DocumentsWriterPerThread.java?rev=1425441&r1=1425440&r2=1425441&view=diff
==============================================================================
--- lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/DocumentsWriterPerThread.java (original)
+++ lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/DocumentsWriterPerThread.java Sun Dec 23 08:49:00 2012
@@ -121,15 +121,21 @@ class DocumentsWriterPerThread {
final SegmentInfoPerCommit segmentInfo;
final FieldInfos fieldInfos;
final FrozenBufferedDeletes segmentDeletes;
+ final BufferedUpdates segmentUpdates;
final MutableBits liveDocs;
+ final UpdatedSegmentData liveUpdates;
final int delCount;
private FlushedSegment(SegmentInfoPerCommit segmentInfo, FieldInfos fieldInfos,
- BufferedDeletes segmentDeletes, MutableBits liveDocs, int delCount) {
+ BufferedDeletes segmentDeletes, MutableBits liveDocs,
+ int delCount, BufferedUpdates segmentUpdates,
+ UpdatedSegmentData liveUpdates) {
this.segmentInfo = segmentInfo;
this.fieldInfos = fieldInfos;
- this.segmentDeletes = segmentDeletes != null && segmentDeletes.any() ? new FrozenBufferedDeletes(segmentDeletes, true) : null;
+ this.segmentDeletes = segmentDeletes != null && segmentDeletes.any() ? new FrozenBufferedDeletes(segmentDeletes, segmentUpdates, true) : null;
+ this.segmentUpdates = segmentUpdates;
this.liveDocs = liveDocs;
+ this.liveUpdates = liveUpdates;
this.delCount = delCount;
}
}
@@ -169,13 +175,16 @@ class DocumentsWriterPerThread {
final TrackingDirectoryWrapper directory;
final Directory directoryOrig;
final DocState docState;
- final DocConsumer consumer;
+ final IndexingChain indexingChain;
final Counter bytesUsed;
+ DocConsumer consumer;
SegmentWriteState flushState;
//Deletes for our still-in-RAM (to be flushed next) segment
BufferedDeletes pendingDeletes;
+ BufferedUpdates pendingUpdates;
SegmentInfo segmentInfo; // Current segment we are working on
+ SegmentInfo baseSegmentInfo; // name of the base segment for segmentInfo
boolean aborting = false; // True if an abort is pending
boolean hasAborted = false; // True if the last exception throws by #updateDocument was aborting
@@ -204,11 +213,10 @@ class DocumentsWriterPerThread {
bytesUsed = Counter.newCounter();
byteBlockAllocator = new DirectTrackingAllocator(bytesUsed);
pendingDeletes = new BufferedDeletes();
+ pendingUpdates = new BufferedUpdates();
intBlockAllocator = new IntBlockAllocator(bytesUsed);
initialize();
- // this should be the last call in the ctor
- // it really sucks that we need to pull this within the ctor and pass this ref to the chain!
- consumer = indexingChain.getChain(this);
+ this.indexingChain = indexingChain;
}
public DocumentsWriterPerThread(DocumentsWriterPerThread other, FieldInfos.Builder fieldInfos) {
@@ -239,7 +247,7 @@ class DocumentsWriterPerThread {
docState.analyzer = analyzer;
docState.docID = numDocsInRAM;
if (segmentInfo == null) {
- initSegmentInfo();
+ initSegmentInfo(null, -1);
}
if (INFO_VERBOSE && infoStream.isEnabled("DWPT")) {
infoStream.message("DWPT", Thread.currentThread().getName() + " update delTerm=" + delTerm + " docID=" + docState.docID + " seg=" + segmentInfo.name);
@@ -272,17 +280,27 @@ class DocumentsWriterPerThread {
abort();
}
}
- finishDocument(delTerm);
+ finishDocument(delTerm, null);
}
- private void initSegmentInfo() {
- String segment = writer.newSegmentName();
- segmentInfo = new SegmentInfo(directoryOrig, Constants.LUCENE_MAIN_VERSION, segment, -1,
- false, codec, null, null);
+ void initSegmentInfo(SegmentInfo info, long updateGen) {
+ if (info == null) {
+ String segment = writer.newSegmentName();
+ segmentInfo = new SegmentInfo(directoryOrig,
+ Constants.LUCENE_MAIN_VERSION, segment, -1, false, codec, null, null);
+ baseSegmentInfo = null;
+ } else {
+ baseSegmentInfo = info;
+ segmentInfo = new SegmentInfo(directoryOrig,
+ Constants.LUCENE_MAIN_VERSION, IndexFileNames.fileNameFromGeneration(
+ info.name, "", updateGen, true), -1, false, codec, null, null);
+ }
assert numDocsInRAM == 0;
if (INFO_VERBOSE && infoStream.isEnabled("DWPT")) {
- infoStream.message("DWPT", Thread.currentThread().getName() + " init seg=" + segment + " delQueue=" + deleteQueue);
+ infoStream.message("DWPT", Thread.currentThread().getName() + " init seg=" + segmentInfo.name + " delQueue=" + deleteQueue);
}
+ // reset consumer, may have previous segment name as inner state
+ consumer = indexingChain.getChain(this);
}
public int updateDocuments(Iterable<? extends IndexDocument> docs, Analyzer analyzer, Term delTerm) throws IOException {
@@ -290,7 +308,7 @@ class DocumentsWriterPerThread {
assert deleteQueue != null;
docState.analyzer = analyzer;
if (segmentInfo == null) {
- initSegmentInfo();
+ initSegmentInfo(null, -1);
}
if (INFO_VERBOSE && infoStream.isEnabled("DWPT")) {
infoStream.message("DWPT", Thread.currentThread().getName() + " update delTerm=" + delTerm + " docID=" + docState.docID + " seg=" + segmentInfo.name);
@@ -330,7 +348,7 @@ class DocumentsWriterPerThread {
}
}
- finishDocument(null);
+ finishDocument(null, null);
}
allDocsIndexed = true;
@@ -338,9 +356,9 @@ class DocumentsWriterPerThread {
// succeeded, but apply it only to docs prior to when
// this batch started:
if (delTerm != null) {
- deleteQueue.add(delTerm, deleteSlice);
+ deleteQueue.add(delTerm, deleteSlice, null);
assert deleteSlice.isTailItem(delTerm) : "expected the delete term as the tail item";
- deleteSlice.apply(pendingDeletes, numDocsInRAM-docCount);
+ deleteSlice.apply(pendingDeletes, pendingUpdates, numDocsInRAM-docCount);
}
} finally {
@@ -360,7 +378,7 @@ class DocumentsWriterPerThread {
return docCount;
}
- private void finishDocument(Term delTerm) {
+ private void finishDocument(Term delTerm, FieldsUpdate fieldsUpdate) {
/*
* here we actually finish the document in two steps 1. push the delete into
* the queue and update our slice. 2. increment the DWPT private document
@@ -372,20 +390,21 @@ class DocumentsWriterPerThread {
if (deleteSlice == null) {
deleteSlice = deleteQueue.newSlice();
if (delTerm != null) {
- deleteQueue.add(delTerm, deleteSlice);
+ deleteQueue.add(delTerm, deleteSlice, fieldsUpdate);
deleteSlice.reset();
}
-
} else {
if (delTerm != null) {
- deleteQueue.add(delTerm, deleteSlice);
+ deleteQueue.add(delTerm, deleteSlice, fieldsUpdate);
assert deleteSlice.isTailItem(delTerm) : "expected the delete term as the tail item";
- deleteSlice.apply(pendingDeletes, numDocsInRAM);
+ deleteSlice.apply(pendingDeletes, pendingUpdates, numDocsInRAM);
} else if (deleteQueue.updateSlice(deleteSlice)) {
- deleteSlice.apply(pendingDeletes, numDocsInRAM);
+ deleteSlice.apply(pendingDeletes, pendingUpdates, numDocsInRAM);
}
}
- ++numDocsInRAM;
+ if (fieldsUpdate == null) {
+ ++numDocsInRAM;
+ }
}
// Buffer a specific docID for deletion. Currently only
@@ -422,7 +441,9 @@ class DocumentsWriterPerThread {
/** Reset after a flush */
private void doAfterFlush() {
segmentInfo = null;
- consumer.doAfterFlush();
+ if (consumer != null) {
+ consumer.doAfterFlush();
+ }
directory.getCreatedFiles().clear();
fieldInfos = new FieldInfos.Builder(fieldInfos.globalFieldNumbers);
parent.subtractFlushedNumDocs(numDocsInRAM);
@@ -441,7 +462,7 @@ class DocumentsWriterPerThread {
adding a document. */
if (deleteSlice != null) {
// apply all deletes before we flush and release the delete slice
- deleteSlice.apply(pendingDeletes, numDocsInRAM);
+ deleteSlice.apply(pendingDeletes, pendingUpdates, numDocsInRAM);
assert deleteSlice.isEmpty();
deleteSlice = null;
}
@@ -449,13 +470,17 @@ class DocumentsWriterPerThread {
}
/** Flush all pending docs to a new segment */
- FlushedSegment flush() throws IOException {
+ FlushedSegment flush(long updateGen) throws IOException {
assert numDocsInRAM > 0;
assert deleteSlice == null : "all deletes must be applied in prepareFlush";
+ if (segmentInfo == null) {
+ return null;
+ }
segmentInfo.setDocCount(numDocsInRAM);
- flushState = new SegmentWriteState(infoStream, directory, segmentInfo, fieldInfos.finish(),
- writer.getConfig().getTermIndexInterval(),
- pendingDeletes, new IOContext(new FlushInfo(numDocsInRAM, bytesUsed())));
+ IOContext context = new IOContext(new FlushInfo(numDocsInRAM, bytesUsed()));
+ flushState = new SegmentWriteState(infoStream, directory, segmentInfo, 0, fieldInfos.finish(),
+ writer.getConfig().getTermIndexInterval(),
+ pendingDeletes, pendingUpdates, context);
final double startMBUsed = parent.flushControl.netBytes() / 1024. / 1024.;
// Apply delete-by-docID now (delete-byDocID only
@@ -487,9 +512,14 @@ class DocumentsWriterPerThread {
try {
consumer.flush(flushState);
pendingDeletes.terms.clear();
- segmentInfo.setFiles(new HashSet<String>(directory.getCreatedFiles()));
-
- final SegmentInfoPerCommit segmentInfoPerCommit = new SegmentInfoPerCommit(segmentInfo, 0, -1L);
+ if (updateGen < 0) {
+ segmentInfo.setFiles(new HashSet<String>(directory.getCreatedFiles()));
+ } else {
+ segmentInfo = baseSegmentInfo;
+ segmentInfo.addFiles(new HashSet<String>(directory.getCreatedFiles()));
+ }
+
+ final SegmentInfoPerCommit segmentInfoPerCommit = new SegmentInfoPerCommit(segmentInfo, 0, -1L, updateGen);
if (infoStream.isEnabled("DWPT")) {
infoStream.message("DWPT", "new segment has " + (flushState.liveDocs == null ? 0 : (flushState.segmentInfo.getDocCount() - flushState.delCountOnFlush)) + " deleted docs");
infoStream.message("DWPT", "new segment has " +
@@ -524,8 +554,9 @@ class DocumentsWriterPerThread {
assert segmentInfo != null;
FlushedSegment fs = new FlushedSegment(segmentInfoPerCommit, flushState.fieldInfos,
- segmentDeletes, flushState.liveDocs, flushState.delCountOnFlush);
- sealFlushedSegment(fs);
+ segmentDeletes, flushState.liveDocs, flushState.delCountOnFlush,
+ pendingUpdates, flushState.liveUpdates);
+ sealFlushedSegment(fs, updateGen);
doAfterFlush();
success = true;
@@ -544,7 +575,7 @@ class DocumentsWriterPerThread {
* Seals the {@link SegmentInfo} for the new flushed segment and persists
* the deleted documents {@link MutableBits}.
*/
- void sealFlushedSegment(FlushedSegment flushedSegment) throws IOException {
+ void sealFlushedSegment(FlushedSegment flushedSegment, long updateGen) throws IOException {
assert flushedSegment != null;
SegmentInfoPerCommit newSegment = flushedSegment.segmentInfo;
@@ -558,7 +589,7 @@ class DocumentsWriterPerThread {
if (writer.useCompoundFile(newSegment)) {
// Now build compound file
- Collection<String> oldFiles = IndexWriter.createCompoundFile(infoStream, directory, MergeState.CheckAbort.NONE, newSegment.info, context);
+ Collection<String> oldFiles = IndexWriter.createCompoundFile(infoStream, directory, MergeState.CheckAbort.NONE, newSegment.info, context, updateGen);
newSegment.info.setUseCompoundFile(true);
writer.deleteNewFiles(oldFiles);
}
@@ -567,7 +598,11 @@ class DocumentsWriterPerThread {
// creating CFS so that 1) .si isn't slurped into CFS,
// and 2) .si reflects useCompoundFile=true change
// above:
- codec.segmentInfoFormat().getSegmentInfoWriter().write(directory, newSegment.info, flushedSegment.fieldInfos, context);
+ if (updateGen < 0) {
+ codec.segmentInfoFormat().getSegmentInfoWriter().write(directory, newSegment.info, flushedSegment.fieldInfos, context);
+ } else {
+ codec.segmentInfoFormat().getSegmentInfoWriter().writeFilesList(directory, newSegment.info, updateGen, context);
+ }
// TODO: ideally we would freeze newSegment here!!
// because any changes after writing the .si will be
@@ -651,9 +686,9 @@ class DocumentsWriterPerThread {
}
}
- PerDocWriteState newPerDocWriteState(String segmentSuffix) {
+ PerDocWriteState newPerDocWriteState() {
assert segmentInfo != null;
- return new PerDocWriteState(infoStream, directory, segmentInfo, bytesUsed, segmentSuffix, IOContext.DEFAULT);
+ return new PerDocWriteState(infoStream, directory, segmentInfo, bytesUsed, "", IOContext.DEFAULT);
}
@Override
@@ -662,4 +697,15 @@ class DocumentsWriterPerThread {
+ ", segment=" + (segmentInfo != null ? segmentInfo.name : "null") + ", aborting=" + aborting + ", numDocsInRAM="
+ numDocsInRAM + ", deleteQueue=" + deleteQueue + "]";
}
+
+ void updateFields(Term term, FieldsUpdate fieldUpdates) {
+ finishDocument(term, fieldUpdates);
+ }
+
+ void clearDeleteSlice() {
+ if (deleteSlice != null) {
+ assert deleteSlice.sliceHead == deleteSlice.sliceTail;
+ deleteSlice = null;
+ }
+ }
}
Added: lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/FieldGenerationReplacements.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/FieldGenerationReplacements.java?rev=1425441&view=auto
==============================================================================
--- lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/FieldGenerationReplacements.java (added)
+++ lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/FieldGenerationReplacements.java Sun Dec 23 08:49:00 2012
@@ -0,0 +1,83 @@
+package org.apache.lucene.index;
+
+import java.util.Iterator;
+import java.util.Map.Entry;
+import java.util.TreeMap;
+
+/*
+ * 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.
+ */
+
+/**
+ * Holds for a certain field in a stacked segment the documents document the
+ * generation in which the last replacement (of the relevant field) took place.
+ */
+public class FieldGenerationReplacements implements Iterable<Entry<Integer,Long>> {
+
+ TreeMap<Integer,Long> map = null;
+
+ /**
+ * Set the generation value for a given document.
+ *
+ * @param docId
+ * Document id.
+ * @param generation
+ * The requested generation.
+ */
+ public void set(int docId, long generation) {
+ if (map == null) {
+ map = new TreeMap<Integer,Long>();
+ }
+ assert generation > 0 && generation <= Integer.MAX_VALUE;
+ map.put(docId, generation);
+ }
+
+ /**
+ * Get the generation value for a given document.
+ *
+ * @param docId
+ * Document id.
+ * @return The requested generation, or -1 if the document has no generation.
+ */
+ public long get(int docId) {
+ if (map == null) {
+ return -1;
+ }
+ final Long val = map.get(docId);
+ if (val == null) {
+ return -1;
+ }
+ return val;
+ }
+
+ public void merge(FieldGenerationReplacements other) {
+ if (map == null) {
+ map = other.map;
+ } else if (other != null) {
+ map.putAll(other.map);
+ }
+ }
+
+ @Override
+ public Iterator<Entry<Integer,Long>> iterator() {
+ return map.entrySet().iterator();
+ }
+
+ public int size() {
+ return map.size();
+ }
+
+}
Added: lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/FieldsUpdate.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/FieldsUpdate.java?rev=1425441&view=auto
==============================================================================
--- lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/FieldsUpdate.java (added)
+++ lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/FieldsUpdate.java Sun Dec 23 08:49:00 2012
@@ -0,0 +1,86 @@
+package org.apache.lucene.index;
+
+import org.apache.lucene.analysis.Analyzer;
+
+/*
+ * 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.
+ */
+
+public class FieldsUpdate implements Comparable<FieldsUpdate> {
+
+ /**
+ * Specifies the operation to perform when updating fields.
+ */
+ enum Operation {
+ /**
+ * Add the given fields to all existing documents matching the update
+ * criterion.
+ */
+ ADD_FIELDS,
+
+ /**
+ * Use the given fields to replace fields with same names in all existing
+ * documents matching the update criterion.
+ */
+ REPLACE_FIELDS
+ }
+
+ final Operation operation;
+ final IndexDocument fields;
+ final Analyzer analyzer;
+ final int docIDUpto;
+
+ /**
+ * An update of fields which is not assigned to a specific live segment.
+ *
+ * @param operation
+ * The type of update operation.
+ * @param fields
+ * The fields to use in the update.
+ * @param analyzer
+ * The analyzer to use in the update.
+ */
+ public FieldsUpdate(Operation operation, IndexDocument fields,
+ Analyzer analyzer) {
+ this.operation = operation;
+ this.fields = fields;
+ this.analyzer = analyzer;
+ this.docIDUpto = -1;
+ }
+
+ /**
+ * An update of fields for a specific live segment.
+ *
+ * @param other
+ * A non-specific update with the update data.
+ * @param docIDUpto
+ * The doc ID in the live segment up to which the update should be
+ * applied.
+ */
+ public FieldsUpdate(FieldsUpdate other, int docIDUpto) {
+ this.operation = other.operation;
+ this.fields = other.fields;
+ this.analyzer = other.analyzer;
+ this.docIDUpto = docIDUpto;
+ }
+
+ /* Order FrieldsUpdate by increasing docIDUpto */
+ @Override
+ public int compareTo(FieldsUpdate other) {
+ return this.docIDUpto - other.docIDUpto;
+ }
+
+}
Modified: lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/FreqProxTermsWriterPerField.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/FreqProxTermsWriterPerField.java?rev=1425441&r1=1425440&r2=1425441&view=diff
==============================================================================
--- lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/FreqProxTermsWriterPerField.java (original)
+++ lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/FreqProxTermsWriterPerField.java Sun Dec 23 08:49:00 2012
@@ -19,7 +19,9 @@ package org.apache.lucene.index;
import java.io.IOException;
import java.util.Comparator;
+import java.util.Iterator;
import java.util.Map;
+import java.util.SortedSet;
import org.apache.lucene.analysis.tokenattributes.OffsetAttribute;
import org.apache.lucene.analysis.tokenattributes.PayloadAttribute;
@@ -78,7 +80,6 @@ final class FreqProxTermsWriterPerField
@Override
void skippingLongTerm() {}
- @Override
public int compareTo(FreqProxTermsWriterPerField other) {
return fieldInfo.name.compareTo(other.fieldInfo.name);
}
@@ -365,7 +366,14 @@ final class FreqProxTermsWriterPerField
} else {
segDeletes = null;
}
-
+
+ final Map<Term,SortedSet<FieldsUpdate>> segUpdates;
+ if (state.segUpdates != null && state.segUpdates.terms.size() > 0) {
+ segUpdates = state.segUpdates.terms;
+ } else {
+ segUpdates = null;
+ }
+
final int[] termIDs = termsHashPerField.sortPostings(termComp);
final int numTerms = termsHashPerField.bytesHash.size();
final BytesRef text = new BytesRef();
@@ -398,6 +406,8 @@ final class FreqProxTermsWriterPerField
final PostingsConsumer postingsConsumer = termsConsumer.startTerm(text);
+ Term term = new Term(fieldName, text);
+
final int delDocLimit;
if (segDeletes != null) {
protoTerm.bytes = text;
@@ -411,6 +421,19 @@ final class FreqProxTermsWriterPerField
delDocLimit = 0;
}
+ final SortedSet<FieldsUpdate> termUpdates;
+ Iterator<FieldsUpdate> updatesIterator = null;
+ FieldsUpdate nextUpdate = null;
+ if (segUpdates != null) {
+ termUpdates = segUpdates.get(term);
+ if (termUpdates != null && !termUpdates.isEmpty()) {
+ updatesIterator = termUpdates.iterator();
+ nextUpdate = updatesIterator.next();
+ }
+ } else {
+ termUpdates = null;
+ }
+
// Now termStates has numToMerge FieldMergeStates
// which all share the same term. Now we must
// interleave the docID streams.
@@ -483,6 +506,23 @@ final class FreqProxTermsWriterPerField
}
}
+ // make sure we update the relevant documents according to the doc ID
+ // in which the updates arrived
+ while (nextUpdate != null && docID > nextUpdate.docIDUpto) {
+ if (updatesIterator.hasNext()) {
+ nextUpdate = updatesIterator.next();
+ } else {
+ nextUpdate = null;
+ }
+ }
+
+ if (nextUpdate != null) {
+ if (state.liveUpdates == null) {
+ state.liveUpdates = new UpdatedSegmentData();
+ }
+ state.liveUpdates.addUpdate(docID, nextUpdate);
+ }
+
totTF += termFreq;
// Carefully copy over the prox + payload info,
Modified: lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/FrozenBufferedDeletes.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/FrozenBufferedDeletes.java?rev=1425441&r1=1425440&r2=1425441&view=diff
==============================================================================
--- lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/FrozenBufferedDeletes.java (original)
+++ lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/FrozenBufferedDeletes.java Sun Dec 23 08:49:00 2012
@@ -17,8 +17,12 @@ package org.apache.lucene.index;
* limitations under the License.
*/
+import java.util.ArrayList;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
+import java.util.SortedSet;
import org.apache.lucene.search.Query;
import org.apache.lucene.util.ArrayUtil;
@@ -35,6 +39,7 @@ class FrozenBufferedDeletes {
/* Query we often undercount (say 24 bytes), plus int. */
final static int BYTES_PER_DEL_QUERY = RamUsageEstimator.NUM_BYTES_OBJECT_REF + RamUsageEstimator.NUM_BYTES_INT + 24;
+ final static List<Term> EMPTY_LIST = new ArrayList<Term>(0);
// Terms, in sorted order:
final PrefixCodedTerms terms;
int termCount; // just for debugging
@@ -49,31 +54,68 @@ class FrozenBufferedDeletes {
final boolean isSegmentPrivate; // set to true iff this frozen packet represents
// a segment private deletes. in that case is should
// only have Queries
+
+ // Updated terms, in sorted order:
+ final PrefixCodedTerms updateTerms;
+ // Updated fields per term
+ final FieldsUpdate[][] updateArrays;
-
- public FrozenBufferedDeletes(BufferedDeletes deletes, boolean isSegmentPrivate) {
+ public FrozenBufferedDeletes(BufferedDeletes deletes, BufferedUpdates updates, boolean isSegmentPrivate) {
this.isSegmentPrivate = isSegmentPrivate;
- assert !isSegmentPrivate || deletes.terms.size() == 0 : "segment private package should only have del queries";
- Term termsArray[] = deletes.terms.keySet().toArray(new Term[deletes.terms.size()]);
- termCount = termsArray.length;
- ArrayUtil.mergeSort(termsArray);
- PrefixCodedTerms.Builder builder = new PrefixCodedTerms.Builder();
- for (Term term : termsArray) {
- builder.add(term);
+ int localBytesUsed = 0;
+ if (deletes != null) {
+ assert !isSegmentPrivate || deletes.terms.size() == 0 : "segment private package should only have del queries";
+ Term termsArray[] = deletes.terms.keySet().toArray(
+ new Term[deletes.terms.size()]);
+ termCount = termsArray.length;
+ ArrayUtil.mergeSort(termsArray);
+ PrefixCodedTerms.Builder builder = new PrefixCodedTerms.Builder();
+ for (Term term : termsArray) {
+ builder.add(term);
+ }
+ terms = builder.finish();
+ localBytesUsed += (int) terms.getSizeInBytes();
+
+ queries = new Query[deletes.queries.size()];
+ queryLimits = new int[deletes.queries.size()];
+ int upto = 0;
+ for (Map.Entry<Query,Integer> ent : deletes.queries.entrySet()) {
+ queries[upto] = ent.getKey();
+ queryLimits[upto] = ent.getValue();
+ upto++;
+ }
+
+ localBytesUsed += queries.length * BYTES_PER_DEL_QUERY;
+ numTermDeletes = deletes.numTermDeletes.get();
+ } else {
+ terms = null;
+ numTermDeletes = 0;
+ queries = null;
+ queryLimits = null;
}
- terms = builder.finish();
- queries = new Query[deletes.queries.size()];
- queryLimits = new int[deletes.queries.size()];
- int upto = 0;
- for(Map.Entry<Query,Integer> ent : deletes.queries.entrySet()) {
- queries[upto] = ent.getKey();
- queryLimits[upto] = ent.getValue();
- upto++;
+ // freeze updates
+ if (updates != null && !updates.terms.isEmpty()) {
+ PrefixCodedTerms.Builder builder = new PrefixCodedTerms.Builder();
+ updateArrays = new FieldsUpdate[updates.terms.size()][];
+ localBytesUsed += RamUsageEstimator.NUM_BYTES_OBJECT_REF * (1 + updateArrays.length);
+ int i = 0;
+ for (Entry<Term,SortedSet<FieldsUpdate>> entry : updates.terms.entrySet()) {
+ builder.add(entry.getKey());
+ SortedSet<FieldsUpdate> updateList = entry.getValue();
+ // TODO : calculate bytes of updates?
+ updateArrays[i] = updateList.toArray(new FieldsUpdate[updateList.size()]);
+ localBytesUsed += RamUsageEstimator.NUM_BYTES_OBJECT_REF * (1 + updateArrays[i].length);
+ i++;
+ }
+ updateTerms = builder.finish();
+ localBytesUsed += (int) updateTerms.getSizeInBytes();
+ } else {
+ updateTerms = null;
+ updateArrays = null;
}
-
- bytesUsed = (int) terms.getSizeInBytes() + queries.length * BYTES_PER_DEL_QUERY;
- numTermDeletes = deletes.numTermDeletes.get();
+
+ bytesUsed = localBytesUsed;
}
public void setDelGen(long gen) {
@@ -90,6 +132,9 @@ class FrozenBufferedDeletes {
return new Iterable<Term>() {
@Override
public Iterator<Term> iterator() {
+ if (terms == null) {
+ return EMPTY_LIST.iterator();
+ }
return terms.iterator();
}
};
@@ -129,9 +174,12 @@ class FrozenBufferedDeletes {
if (numTermDeletes != 0) {
s += " " + numTermDeletes + " deleted terms (unique count=" + termCount + ")";
}
- if (queries.length != 0) {
+ if (queries != null && queries.length != 0) {
s += " " + queries.length + " deleted queries";
}
+ if (updateArrays != null && updateArrays.length > 0) {
+ s += " " + updateArrays.length + " updates";
+ }
if (bytesUsed != 0) {
s += " bytesUsed=" + bytesUsed;
}
@@ -139,7 +187,11 @@ class FrozenBufferedDeletes {
return s;
}
- boolean any() {
- return termCount > 0 || queries.length > 0;
+ boolean anyDeletes() {
+ return termCount > 0 || (queries != null && queries.length > 0);
+ }
+
+ boolean anyUpdates() {
+ return updateTerms != null;
}
}
Modified: lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/IndexFileNames.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/IndexFileNames.java?rev=1425441&r1=1425440&r2=1425441&view=diff
==============================================================================
--- lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/IndexFileNames.java (original)
+++ lucene/dev/branches/lucene4258/lucene/core/src/java/org/apache/lucene/index/IndexFileNames.java Sun Dec 23 08:49:00 2012
@@ -29,7 +29,7 @@ import org.apache.lucene.codecs.Codec;
* name matches an extension ({@link #matchesExtension(String, String)
* matchesExtension}), as well as generating file names from a segment name,
* generation and extension (
- * {@link #fileNameFromGeneration(String, String, long) fileNameFromGeneration},
+ * {@link #fileNameFromGeneration(String, String, long, boolean) fileNameFromGeneration},
* {@link #segmentFileName(String, String, String) segmentFileName}).
*
* <p><b>NOTE</b>: extensions used by codecs are not
@@ -83,8 +83,9 @@ public final class IndexFileNames {
* @param base main part of the file name
* @param ext extension of the filename
* @param gen generation
+ * @param isUpdate whether the file is an update file or not
*/
- public static String fileNameFromGeneration(String base, String ext, long gen) {
+ public static String fileNameFromGeneration(String base, String ext, long gen, boolean isUpdate) {
if (gen == -1) {
return null;
} else if (gen == 0) {
@@ -94,8 +95,11 @@ public final class IndexFileNames {
// The '6' part in the length is: 1 for '.', 1 for '_' and 4 as estimate
// to the gen length as string (hopefully an upper limit so SB won't
// expand in the middle.
- StringBuilder res = new StringBuilder(base.length() + 6 + ext.length())
- .append(base).append('_').append(Long.toString(gen, Character.MAX_RADIX));
+ StringBuilder res = new StringBuilder(base.length() + 6 + ext.length());
+ if (isUpdate) {
+ res.append('_');
+ }
+ res.append(base).append('_').append(generationString(gen));
if (ext.length() > 0) {
res.append('.').append(ext);
}
@@ -103,6 +107,35 @@ public final class IndexFileNames {
}
}
+ public static String generationString(long gen) {
+ return Long.toString(gen, Character.MAX_RADIX);
+ }
+
+ /**
+ * Computes the base name of an updated segment from base and generation. If
+ * the generation < 0, the file name is null. otherwise, the file name is
+ * <base>_upd_<gen>.<br>
+ *
+ * @param baseName
+ * base segment string
+ * @param gen
+ * update generation
+ */
+ public static String updatedSegmentFileNameFromGeneration(String baseName,
+ long gen) {
+ if (gen <= 0) {
+ return null;
+ } else {
+ assert gen > 0;
+ // The '10' part in the length is: 3 for '_', 3 for "upd" and 4 as
+ // estimate to the gen length as string (hopefully an upper limit so SB
+ // won't expand in the middle.
+ StringBuilder res = new StringBuilder(baseName.length() + 10).append('_')
+ .append(baseName).append('_').append(generationString(gen));
+ return res.toString();
+ }
+ }
+
/**
* Returns a file name that includes the given segment name, your own custom
* name and extension. The format of the filename is:
@@ -201,5 +234,5 @@ public final class IndexFileNames {
// All files created by codecs much match this pattern (we
// check this in SegmentInfo.java):
- static final Pattern CODEC_FILE_PATTERN = Pattern.compile("_[a-z0-9]+(_.*)?\\..*");
+ static final Pattern CODEC_FILE_PATTERN = Pattern.compile("_[_]?[a-z0-9]+(_.*)?\\..*");
}