You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@lucene.apache.org by GitBox <gi...@apache.org> on 2020/08/18 00:33:55 UTC
[GitHub] [lucene-solr] mikemccand commented on a change in pull request #1623: LUCENE-8962: Merge segments on getReader
mikemccand commented on a change in pull request #1623:
URL: https://github.com/apache/lucene-solr/pull/1623#discussion_r470869880
##########
File path: lucene/core/src/java/org/apache/lucene/index/IndexWriter.java
##########
@@ -607,6 +633,57 @@ DirectoryReader getReader(boolean applyAllDeletes, boolean writeAllDeletes) thro
}
}
}
+ if (onCommitMerges != null) { // only relevant if we do merge on getReader
+ boolean replaceReaderSuccess = false;
Review comment:
Thank you for good `success` variable naming instead of the usual `success1` and `success2` etc.!
##########
File path: lucene/core/src/java/org/apache/lucene/index/IndexWriter.java
##########
@@ -3335,49 +3416,60 @@ public void mergeFinished(boolean committed, boolean segmentDropped) throws IOEx
// includedInCommit will be set (above, by our caller) to false if the allowed max wall clock
// time (IWC.getMaxCommitMergeWaitMillis()) has elapsed, which means we did not make the timeout
// and will not commit our merge to the to-be-commited SegmentInfos
-
if (segmentDropped == false
&& committed
- && includeInCommit.get()) {
+ && includeMergeResult.get()) {
+
+ // make sure onMergeComplete really was called:
+ assert origInfo != null;
if (infoStream.isEnabled("IW")) {
infoStream.message("IW", "now apply merge during commit: " + toWrap.segString());
}
- // make sure onMergeComplete really was called:
- assert origInfo != null;
-
- deleter.incRef(origInfo.files());
+ if (trigger == MergeTrigger.COMMIT) { // if we do this in a getReader call here this is obsolete
+ deleter.incRef(origInfo.files());
Review comment:
Hmm why don't we need to `deleter.incRef` for NRT reader case? I guess the NRT reader we opened holds a reference already?
##########
File path: lucene/core/src/java/org/apache/lucene/index/IndexWriter.java
##########
@@ -630,6 +694,64 @@ DirectoryReader getReader(boolean applyAllDeletes, boolean writeAllDeletes) thro
return r;
}
+ private StandardDirectoryReader finishGetReaderMerge(AtomicBoolean hasTimedOut, Map<String, SegmentReader> mergedReaders,
+ Map<String, SegmentReader> openedReadOnlyClones, SegmentInfos openingSegmentInfos,
+ boolean applyAllDeletes, boolean writeAllDeletes,
+ MergePolicy.MergeSpecification onCommitMerges, long maxCommitMergeWaitMillis) throws IOException {
+ boolean replaceReaderSuccess = false;
+ try {
+ mergeScheduler.merge(mergeSource, MergeTrigger.GET_READER);
+ onCommitMerges.await(maxCommitMergeWaitMillis, TimeUnit.MILLISECONDS);
+ assert openingSegmentInfos != null;
+ synchronized (this) {
+ hasTimedOut.set(true);
Review comment:
Hmm, what if the merges finished before the timeout? The `await` would return early, and return `true` if it did not timeout (I think?). Maybe we do not need/care to distinguish that? In which case maybe renamed `hasTimedOut` to something else (`mergesFinished`?).
##########
File path: lucene/core/src/java/org/apache/lucene/index/StandardDirectoryReader.java
##########
@@ -82,7 +82,8 @@ protected DirectoryReader doBody(String segmentFileName) throws IOException {
}
/** Used by near real-time search */
- static DirectoryReader open(IndexWriter writer, SegmentInfos infos, boolean applyAllDeletes, boolean writeAllDeletes) throws IOException {
+ static StandardDirectoryReader open(IndexWriter writer, IOUtils.IOFunction<SegmentCommitInfo, SegmentReader> readerFunction,
+ SegmentInfos infos, boolean applyAllDeletes, boolean writeAllDeletes) throws IOException {
Review comment:
Re-indent?
##########
File path: lucene/core/src/java/org/apache/lucene/index/IndexWriter.java
##########
@@ -607,6 +633,57 @@ DirectoryReader getReader(boolean applyAllDeletes, boolean writeAllDeletes) thro
}
}
}
+ if (onCommitMerges != null) { // only relevant if we do merge on getReader
+ boolean replaceReaderSuccess = false;
+ try {
+ mergeScheduler.merge(mergeSource, MergeTrigger.GET_READER);
+ onCommitMerges.await(maxCommitMergeWaitMillis, TimeUnit.MILLISECONDS);
+ assert openingSegmentInfos != null;
+ synchronized (this) {
+ includeMergeReader.set(false);
+ boolean openNewReader = mergedReaders.isEmpty() == false;
+ if (openNewReader) {
+ StandardDirectoryReader mergedReader = StandardDirectoryReader.open(this,
+ sci -> {
+ // as soon as we remove the reader and return it the StandardDirectoryReader#open
+ // will take care of closing it. We only need to handle the readers that remain in the
+ // mergedReaders map and close them.
+ SegmentReader remove = mergedReaders.remove(sci.info.name);
+ if (remove == null) {
+ remove = openedReadOnlyClones.remove(sci.info.name);
+ assert remove != null;
+ // each of the readers we reuse from the previous reader needs to be refInced
+ // since we reuse them but don't have an implicit refInc in the SDR:open call
Review comment:
s/`refInc`/`incRef`?
##########
File path: lucene/core/src/java/org/apache/lucene/index/IndexWriter.java
##########
@@ -607,6 +633,57 @@ DirectoryReader getReader(boolean applyAllDeletes, boolean writeAllDeletes) thro
}
}
}
+ if (onCommitMerges != null) { // only relevant if we do merge on getReader
+ boolean replaceReaderSuccess = false;
+ try {
+ mergeScheduler.merge(mergeSource, MergeTrigger.GET_READER);
+ onCommitMerges.await(maxCommitMergeWaitMillis, TimeUnit.MILLISECONDS);
+ assert openingSegmentInfos != null;
+ synchronized (this) {
+ includeMergeReader.set(false);
+ boolean openNewReader = mergedReaders.isEmpty() == false;
+ if (openNewReader) {
+ StandardDirectoryReader mergedReader = StandardDirectoryReader.open(this,
+ sci -> {
+ // as soon as we remove the reader and return it the StandardDirectoryReader#open
+ // will take care of closing it. We only need to handle the readers that remain in the
+ // mergedReaders map and close them.
+ SegmentReader remove = mergedReaders.remove(sci.info.name);
+ if (remove == null) {
+ remove = openedReadOnlyClones.remove(sci.info.name);
+ assert remove != null;
+ // each of the readers we reuse from the previous reader needs to be refInced
Review comment:
s/`refInced`/`incRef'd`?
##########
File path: lucene/core/src/java/org/apache/lucene/index/IndexWriter.java
##########
@@ -3335,49 +3416,60 @@ public void mergeFinished(boolean committed, boolean segmentDropped) throws IOEx
// includedInCommit will be set (above, by our caller) to false if the allowed max wall clock
// time (IWC.getMaxCommitMergeWaitMillis()) has elapsed, which means we did not make the timeout
// and will not commit our merge to the to-be-commited SegmentInfos
-
if (segmentDropped == false
&& committed
- && includeInCommit.get()) {
+ && includeMergeResult.get()) {
+
+ // make sure onMergeComplete really was called:
+ assert origInfo != null;
if (infoStream.isEnabled("IW")) {
infoStream.message("IW", "now apply merge during commit: " + toWrap.segString());
}
- // make sure onMergeComplete really was called:
- assert origInfo != null;
-
- deleter.incRef(origInfo.files());
+ if (trigger == MergeTrigger.COMMIT) { // if we do this in a getReader call here this is obsolete
+ deleter.incRef(origInfo.files());
+ }
Set<String> mergedSegmentNames = new HashSet<>();
for (SegmentCommitInfo sci : segments) {
mergedSegmentNames.add(sci.info.name);
}
List<SegmentCommitInfo> toCommitMergedAwaySegments = new ArrayList<>();
- for (SegmentCommitInfo sci : committingSegmentInfos) {
+ for (SegmentCommitInfo sci : mergingSegmentInfos) {
if (mergedSegmentNames.contains(sci.info.name)) {
toCommitMergedAwaySegments.add(sci);
- deleter.decRef(sci.files());
+ if (trigger == MergeTrigger.COMMIT) { // if we do this in a getReader call here this is obsolete
+ deleter.decRef(sci.files());
+ }
}
}
// Construct a OneMerge that applies to toCommit
MergePolicy.OneMerge applicableMerge = new MergePolicy.OneMerge(toCommitMergedAwaySegments);
applicableMerge.info = origInfo;
long segmentCounter = Long.parseLong(origInfo.info.name.substring(1), Character.MAX_RADIX);
- committingSegmentInfos.counter = Math.max(committingSegmentInfos.counter, segmentCounter + 1);
- committingSegmentInfos.applyMergeChanges(applicableMerge, false);
+ mergingSegmentInfos.counter = Math.max(mergingSegmentInfos.counter, segmentCounter + 1);
+ mergingSegmentInfos.applyMergeChanges(applicableMerge, false);
} else {
if (infoStream.isEnabled("IW")) {
infoStream.message("IW", "skip apply merge during commit: " + toWrap.segString());
}
}
- toWrap.mergeFinished(committed, false);
+ toWrap.mergeFinished(committed, segmentDropped);
super.mergeFinished(committed, segmentDropped);
}
@Override
- void onMergeComplete() {
- // clone the target info to make sure we have the original info without the updated del and update gens
- origInfo = info.clone();
+ void onMergeComplete() throws IOException {
+ if (includeMergeResult.get()
+ && isAborted() == false
+ && info.info.maxDoc() > 0/* never do this if the segment if dropped / empty */) {
+ assert Thread.holdsLock(IndexWriter.this);
Review comment:
Maybe move the `assert` to top of the method? We should always hold IW's monitor lock on entry?
##########
File path: lucene/core/src/java/org/apache/lucene/index/IndexWriter.java
##########
@@ -545,18 +546,54 @@ DirectoryReader getReader(boolean applyAllDeletes, boolean writeAllDeletes) thro
// obtained during this flush are pooled, the first time
// this method is called:
readerPool.enableReaderPooling();
- DirectoryReader r = null;
+ StandardDirectoryReader r = null;
doBeforeFlush();
- boolean anyChanges = false;
+ boolean anyChanges;
/*
* for releasing a NRT reader we must ensure that
* DW doesn't add any segments or deletes until we are
* done with creating the NRT DirectoryReader.
* We release the two stage full flush after we are done opening the
* directory reader!
*/
+ MergePolicy.MergeSpecification onGetReaderMerges = null;
+ AtomicBoolean hasTimedOut = new AtomicBoolean(false);
+ Map<String, SegmentReader> mergedReaders = new HashMap<>();
+ Map<String, SegmentReader> openedReadOnlyClones = new HashMap<>();
+ // this function is used to control which SR are opened in order to keep track of them
+ // and to reuse them in the case we wait for merges in this getReader call.
+ IOUtils.IOFunction<SegmentCommitInfo, SegmentReader> readerFactory = sci -> {
+ final ReadersAndUpdates rld = getPooledInstance(sci, true);
+ try {
+ assert Thread.holdsLock(IndexWriter.this);
+ SegmentReader segmentReader = rld.getReadOnlyClone(IOContext.READ);
+ openedReadOnlyClones.put(sci.info.name, segmentReader);
+ return segmentReader;
+ } finally {
+ release(rld);
+ }
+ };
+ SegmentInfos openingSegmentInfos = null;
+ final long maxFullFlushMergeWaitMillis = config.getMaxFullFlushMergeWaitMillis();
boolean success2 = false;
try {
+ /* this is the essential part of the getReader method. We need to take care of the following things:
Review comment:
Thank you for these awesome details!!
##########
File path: lucene/core/src/java/org/apache/lucene/index/IndexWriter.java
##########
@@ -545,18 +546,54 @@ DirectoryReader getReader(boolean applyAllDeletes, boolean writeAllDeletes) thro
// obtained during this flush are pooled, the first time
// this method is called:
readerPool.enableReaderPooling();
- DirectoryReader r = null;
+ StandardDirectoryReader r = null;
doBeforeFlush();
- boolean anyChanges = false;
+ boolean anyChanges;
/*
* for releasing a NRT reader we must ensure that
* DW doesn't add any segments or deletes until we are
* done with creating the NRT DirectoryReader.
* We release the two stage full flush after we are done opening the
* directory reader!
*/
+ MergePolicy.MergeSpecification onGetReaderMerges = null;
+ AtomicBoolean hasTimedOut = new AtomicBoolean(false);
+ Map<String, SegmentReader> mergedReaders = new HashMap<>();
+ Map<String, SegmentReader> openedReadOnlyClones = new HashMap<>();
+ // this function is used to control which SR are opened in order to keep track of them
+ // and to reuse them in the case we wait for merges in this getReader call.
+ IOUtils.IOFunction<SegmentCommitInfo, SegmentReader> readerFactory = sci -> {
+ final ReadersAndUpdates rld = getPooledInstance(sci, true);
+ try {
+ assert Thread.holdsLock(IndexWriter.this);
+ SegmentReader segmentReader = rld.getReadOnlyClone(IOContext.READ);
+ openedReadOnlyClones.put(sci.info.name, segmentReader);
+ return segmentReader;
+ } finally {
+ release(rld);
+ }
+ };
+ SegmentInfos openingSegmentInfos = null;
+ final long maxFullFlushMergeWaitMillis = config.getMaxFullFlushMergeWaitMillis();
boolean success2 = false;
try {
+ /* this is the essential part of the getReader method. We need to take care of the following things:
+ * - flush all currently in-memory DWPTs to disk
+ * - apply all deletes & updates to new and to the existing DWPTs
+ * - prevent flushes and applying deletes of concurrently indexing DWPTs to be applied
+ * - open a SDR on the updated SIS
+ *
+ * in order do prevent concurrent flushes we call DocumentsWriter#flushAllThreads that swaps out the deleteQueue
Review comment:
s/`do`/`to`?
##########
File path: lucene/core/src/java/org/apache/lucene/index/IndexWriter.java
##########
@@ -3244,7 +3322,7 @@ private long prepareCommitInternal() throws IOException {
if (anyChanges && maxCommitMergeWaitMillis > 0) {
// we can safely call prepareOnCommitMerge since writeReaderPool(true) above wrote all
// necessary files to disk and checkpointed them.
- onCommitMerges = prepareOnCommitMerge(toCommit, includeInCommit);
+ onCommitMerges = preparePointInTimeMerge(toCommit, includeInCommit, MergeTrigger.COMMIT, sci->{});
Review comment:
Good name! Maybe rename `onCommitMerges` to `pointInTimeMerges`?
##########
File path: lucene/core/src/java/org/apache/lucene/index/IndexWriter.java
##########
@@ -630,6 +694,64 @@ DirectoryReader getReader(boolean applyAllDeletes, boolean writeAllDeletes) thro
return r;
}
+ private StandardDirectoryReader finishGetReaderMerge(AtomicBoolean hasTimedOut, Map<String, SegmentReader> mergedReaders,
+ Map<String, SegmentReader> openedReadOnlyClones, SegmentInfos openingSegmentInfos,
+ boolean applyAllDeletes, boolean writeAllDeletes,
+ MergePolicy.MergeSpecification onCommitMerges, long maxCommitMergeWaitMillis) throws IOException {
+ boolean replaceReaderSuccess = false;
+ try {
+ mergeScheduler.merge(mergeSource, MergeTrigger.GET_READER);
+ onCommitMerges.await(maxCommitMergeWaitMillis, TimeUnit.MILLISECONDS);
+ assert openingSegmentInfos != null;
Review comment:
Maybe move this `assert` to top of method?
##########
File path: lucene/core/src/java/org/apache/lucene/index/IndexWriter.java
##########
@@ -607,6 +633,57 @@ DirectoryReader getReader(boolean applyAllDeletes, boolean writeAllDeletes) thro
}
}
}
+ if (onCommitMerges != null) { // only relevant if we do merge on getReader
+ boolean replaceReaderSuccess = false;
+ try {
+ mergeScheduler.merge(mergeSource, MergeTrigger.GET_READER);
+ onCommitMerges.await(maxCommitMergeWaitMillis, TimeUnit.MILLISECONDS);
Review comment:
Maybe we need to rename this `IWC` option and variables? Maybe we need two timeouts, one for `commit`, one for `getReader`? Or, maybe we somehow make this a property of the `MergeSpecification` so `MergePolicy` can decide case by case what the timeout should be?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org