Skip to content

Cut over to unwrap segment reader #33843

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Sep 19, 2018
Merged

Cut over to unwrap segment reader #33843

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions server/src/main/java/org/elasticsearch/common/lucene/Lucene.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
import org.apache.lucene.document.NumericDocValuesField;
import org.apache.lucene.index.CorruptIndexException;
import org.apache.lucene.index.DirectoryReader;
import org.apache.lucene.index.FilterCodecReader;
import org.apache.lucene.index.FilterDirectoryReader;
import org.apache.lucene.index.FilterLeafReader;
import org.apache.lucene.index.IndexCommit;
Expand Down Expand Up @@ -726,6 +727,9 @@ public static SegmentReader segmentReader(LeafReader reader) {
} else if (reader instanceof FilterLeafReader) {
final FilterLeafReader fReader = (FilterLeafReader) reader;
return segmentReader(FilterLeafReader.unwrap(fReader));
} else if (reader instanceof FilterCodecReader) {
final FilterCodecReader fReader = (FilterCodecReader) reader;
return segmentReader(FilterCodecReader.unwrap(fReader));
}
// hard fail - we can't get a SegmentReader
throw new IllegalStateException("Can not extract segment reader from given index reader [" + reader + "]");
Expand Down
83 changes: 13 additions & 70 deletions x-pack/plugin/core/src/main/java/org/elasticsearch/snapshots/SourceOnlySnapshot.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,31 +81,21 @@ public synchronized List<String> syncSnapshot(IndexCommit commit) throws IOExcep
String segmentFileName;
try (Lock writeLock = targetDirectory.obtainLock(IndexWriter.WRITE_LOCK_NAME);
StandardDirectoryReader reader = (StandardDirectoryReader) DirectoryReader.open(commit)) {
SegmentInfos segmentInfos = reader.getSegmentInfos();
SegmentInfos segmentInfos = reader.getSegmentInfos().clone();
DirectoryReader wrappedReader = wrapReader(reader);
List<SegmentCommitInfo> newInfos = new ArrayList<>();
for (LeafReaderContext ctx : reader.leaves()) {
for (LeafReaderContext ctx : wrappedReader.leaves()) {
LeafReader leafReader = ctx.reader();
SegmentCommitInfo info = reader.getSegmentInfos().info(ctx.ord);
assert info.info.equals(Lucene.segmentReader(ctx.reader()).getSegmentInfo().info);
/* We could do this totally different without wrapping this dummy directory reader if FilterCodecReader would have a
* getDelegate method. This is fixed in LUCENE-8502 but we need to wait for it to come in 7.5.1 or 7.6.
* The reason here is that the ctx.ord is not guaranteed to be equivalent to the SegmentCommitInfo ord in the SegmentInfo
* object since we might drop fully deleted segments. if that happens we are using the wrong reader for the SI and
* might almost certainly expose deleted documents.
*/
DirectoryReader wrappedReader = wrapReader(new DummyDirectoryReader(reader.directory(), leafReader));
if (wrappedReader.leaves().isEmpty() == false) {
leafReader = wrappedReader.leaves().get(0).reader();
LiveDocs liveDocs = getLiveDocs(leafReader);
if (leafReader.numDocs() != 0) { // fully deleted segments don't need to be processed
SegmentCommitInfo newInfo = syncSegment(info, liveDocs, leafReader.getFieldInfos(), existingSegments, createdFiles);
newInfos.add(newInfo);
}
SegmentCommitInfo info = Lucene.segmentReader(leafReader).getSegmentInfo();
LiveDocs liveDocs = getLiveDocs(leafReader);
if (leafReader.numDocs() != 0) { // fully deleted segments don't need to be processed
SegmentCommitInfo newInfo = syncSegment(info, liveDocs, leafReader.getFieldInfos(), existingSegments, createdFiles);
newInfos.add(newInfo);
}
}
segmentInfos.clear();
segmentInfos.addAll(newInfos);
segmentInfos.setNextWriteGeneration(Math.max(segmentInfos.getGeneration(), generation)+1);
segmentInfos.setNextWriteGeneration(Math.max(segmentInfos.getGeneration(), generation) + 1);
String pendingSegmentFileName = IndexFileNames.fileNameFromGeneration(IndexFileNames.PENDING_SEGMENTS,
"", segmentInfos.getGeneration());
try (IndexOutput segnOutput = targetDirectory.createOutput(pendingSegmentFileName, IOContext.DEFAULT)) {
Expand Down Expand Up @@ -207,9 +197,9 @@ private SegmentCommitInfo syncSegment(SegmentCommitInfo segmentCommitInfo, LiveD
newInfo = new SegmentCommitInfo(newSegmentInfo, 0, 0, -1, -1, -1);
List<FieldInfo> fieldInfoCopy = new ArrayList<>(fieldInfos.size());
for (FieldInfo fieldInfo : fieldInfos) {
fieldInfoCopy.add(new FieldInfo(fieldInfo.name, fieldInfo.number,
false, false, false, IndexOptions.NONE, DocValuesType.NONE, -1, fieldInfo.attributes(), 0, 0,
fieldInfo.isSoftDeletesField()));
fieldInfoCopy.add(new FieldInfo(fieldInfo.name, fieldInfo.number,
false, false, false, IndexOptions.NONE, DocValuesType.NONE, -1, fieldInfo.attributes(), 0, 0,
fieldInfo.isSoftDeletesField()));
}
FieldInfos newFieldInfos = new FieldInfos(fieldInfoCopy.toArray(new FieldInfo[0]));
codec.fieldInfosFormat().write(trackingDir, newSegmentInfo, segmentSuffix, newFieldInfos, IOContext.DEFAULT);
Expand Down Expand Up @@ -250,7 +240,7 @@ private SegmentCommitInfo syncSegment(SegmentCommitInfo segmentCommitInfo, LiveD

private boolean assertLiveDocs(Bits liveDocs, int deletes) {
int actualDeletes = 0;
for (int i = 0; i < liveDocs.length(); i++ ) {
for (int i = 0; i < liveDocs.length(); i++) {
if (liveDocs.get(i) == false) {
actualDeletes++;
}
Expand All @@ -268,51 +258,4 @@ private static class LiveDocs {
this.bits = bits;
}
}

private static class DummyDirectoryReader extends DirectoryReader {

protected DummyDirectoryReader(Directory directory, LeafReader... segmentReaders) throws IOException {
super(directory, segmentReaders);
}

@Override
protected DirectoryReader doOpenIfChanged() throws IOException {
return null;
}

@Override
protected DirectoryReader doOpenIfChanged(IndexCommit commit) throws IOException {
return null;
}

@Override
protected DirectoryReader doOpenIfChanged(IndexWriter writer, boolean applyAllDeletes) throws IOException {
return null;
}

@Override
public long getVersion() {
return 0;
}

@Override
public boolean isCurrent() throws IOException {
return false;
}

@Override
public IndexCommit getIndexCommit() throws IOException {
return null;
}

@Override
protected void doClose() throws IOException {

}

@Override
public CacheHelper getReaderCacheHelper() {
return null;
}
}
}