Skip to content

Commit

Permalink
Cut over to unwrap segment reader (#33843)
Browse files Browse the repository at this point in the history 
The fix in #33757 introduces some workaround since FilterCodecReader didn't
support unwrapping. This cuts over to a more elegant fix to access the readers
segment infos.
  • Loading branch information
@s1monw
s1monw committed Sep 19, 2018
1 parent 4c6cb3c commit dadc717
Show file tree
Hide file tree
Showing 2 changed files with 17 additions and 70 deletions.
4 changes: 4 additions & 0 deletions server/src/main/java/org/elasticsearch/common/lucene/Lucene.java
View file
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 @@ -668,6 +669,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
Original file line number Diff line number Diff line change
Expand Up @@ -80,31 +80,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 @@ -206,9 +196,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 @@ -249,7 +239,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 @@ -267,51 +257,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;
}
}
}

0 comments on commit dadc717

Please sign in to comment.