Skip to content
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.

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

Allow _update and upsert to read from the transaction log #29264

Merged
merged 5 commits into from
Mar 28, 2018
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -112,13 +112,13 @@ protected ExplainResponse shardOperation(ExplainRequest request, ShardId shardId
if (uidTerm == null) {
return new ExplainResponse(shardId.getIndexName(), request.type(), request.id(), false);
}
result = context.indexShard().get(new Engine.Get(false, request.type(), request.id(), uidTerm));
result = context.indexShard().get(new Engine.Get(false, false, request.type(), request.id(), uidTerm));
if (!result.exists()) {
return new ExplainResponse(shardId.getIndexName(), request.type(), request.id(), false);
}
context.parsedQuery(context.getQueryShardContext().toQuery(request.query()));
context.preProcess(true);
int topLevelDocId = result.docIdAndVersion().docId + result.docIdAndVersion().context.docBase;
int topLevelDocId = result.docIdAndVersion().docId + result.docIdAndVersion().docBase;
Explanation explanation = context.searcher().explain(context.query(), topLevelDocId);
for (RescoreContext ctx : context.rescore()) {
Rescorer rescorer = ctx.rescorer();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,6 @@
import org.elasticsearch.script.ExecutableScript;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptService;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.elasticsearch.search.lookup.SourceLookup;

import java.io.IOException;
Expand All @@ -71,9 +70,8 @@ public UpdateHelper(Settings settings, ScriptService scriptService) {
* Prepares an update request by converting it into an index or delete request or an update response (no action).
*/
public Result prepare(UpdateRequest request, IndexShard indexShard, LongSupplier nowInMillis) {
final GetResult getResult = indexShard.getService().get(request.type(), request.id(),
new String[]{RoutingFieldMapper.NAME, ParentFieldMapper.NAME},
true, request.version(), request.versionType(), FetchSourceContext.FETCH_SOURCE);
final GetResult getResult = indexShard.getService().getForUpdate(request.type(), request.id(), request.version(),
request.versionType());
return prepare(indexShard.shardId(), request, getResult, nowInMillis);
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -100,7 +100,7 @@ public DocIdAndVersion lookupVersion(BytesRef id, LeafReaderContext context)
if (versions.advanceExact(docID) == false) {
throw new IllegalArgumentException("Document [" + docID + "] misses the [" + VersionFieldMapper.NAME + "] field");
}
return new DocIdAndVersion(docID, versions.longValue(), context);
return new DocIdAndVersion(docID, versions.longValue(), context.reader(), context.docBase);
} else {
return null;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.elasticsearch.common.lucene.uid;

import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.LeafReader;
import org.apache.lucene.index.LeafReaderContext;
import org.apache.lucene.index.NumericDocValues;
import org.apache.lucene.index.Term;
Expand Down Expand Up @@ -97,12 +98,14 @@ private VersionsAndSeqNoResolver() {
public static class DocIdAndVersion {
public final int docId;
public final long version;
public final LeafReaderContext context;
public final LeafReader reader;
public final int docBase;

DocIdAndVersion(int docId, long version, LeafReaderContext context) {
public DocIdAndVersion(int docId, long version, LeafReader reader, int docBase) {
this.docId = docId;
this.version = version;
this.context = context;
this.reader = reader;
this.docBase = docBase;
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1232,14 +1232,16 @@ public static class Get {
private final boolean realtime;
private final Term uid;
private final String type, id;
private final boolean readFromTranslog;
private long version = Versions.MATCH_ANY;
private VersionType versionType = VersionType.INTERNAL;

public Get(boolean realtime, String type, String id, Term uid) {
public Get(boolean realtime, boolean readFromTranslog, String type, String id, Term uid) {
this.realtime = realtime;
this.type = type;
this.id = id;
this.uid = uid;
this.readFromTranslog = readFromTranslog;
}

public boolean realtime() {
Expand Down Expand Up @@ -1275,6 +1277,10 @@ public Get versionType(VersionType versionType) {
this.versionType = versionType;
return this;
}

public boolean isReadFromTranslog() {
return readFromTranslog;
}
}

public static class GetResult implements Releasable {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,7 @@
import org.elasticsearch.threadpool.ThreadPool;

import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
Expand Down Expand Up @@ -145,6 +146,7 @@ public class InternalEngine extends Engine {
* being indexed/deleted.
*/
private final AtomicLong writingBytes = new AtomicLong();
private final AtomicBoolean trackTranslogLocation = new AtomicBoolean(false);

@Nullable
private final String historyUUID;
Expand Down Expand Up @@ -558,6 +560,24 @@ public GetResult get(Get get, BiFunction<String, SearcherScope, Searcher> search
throw new VersionConflictEngineException(shardId, get.type(), get.id(),
get.versionType().explainConflictForReads(versionValue.version, get.version()));
}
if (get.isReadFromTranslog()) {
// this is only used for updates - API _GET calls will always read form a reader for consistency
// the update call doesn't need the consistency since it's source only + _parent but parent can go away in 7.0
if (versionValue.getLocation() != null) {
try {
Translog.Operation operation = translog.readOperation(versionValue.getLocation());
if (operation != null) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

when do we expect this to be null?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Given the translog generation is not available anymore I think it’s unlikely but I see a chance. I can add a comment in the code

TranslogLeafReader reader = new TranslogLeafReader((Translog.Index) operation);
return new GetResult(new Searcher("realtime_get", new IndexSearcher(reader)),
new VersionsAndSeqNoResolver.DocIdAndVersion(0, ((Translog.Index) operation).version(), reader, 0));
}
} catch (IOException e) {
throw new UncheckedIOException(e);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we seem to be using EngineException as a wrapper in other places?

}
} else {
trackTranslogLocation.set(true);
}
}
refresh("realtime_get", SearcherScope.INTERNAL);
}
scope = SearcherScope.INTERNAL;
Expand Down Expand Up @@ -790,6 +810,10 @@ public IndexResult index(Index index) throws IOException {
}
indexResult.setTranslogLocation(location);
}
if (plan.indexIntoLucene && indexResult.hasFailure() == false) {
versionMap.maybePutUnderLock(index.uid().bytes(),
getVersionValue(plan.versionForIndexing, plan.seqNoForIndexing, index.primaryTerm(), indexResult.getTranslogLocation()));
}
if (indexResult.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
localCheckpointTracker.markSeqNoAsCompleted(indexResult.getSeqNo());
}
Expand Down Expand Up @@ -916,8 +940,6 @@ private IndexResult indexIntoLucene(Index index, IndexingStrategy plan)
assert assertDocDoesNotExist(index, canOptimizeAddDocument(index) == false);
index(index.docs(), indexWriter);
}
versionMap.maybePutUnderLock(index.uid().bytes(),
new VersionValue(plan.versionForIndexing, plan.seqNoForIndexing, index.primaryTerm()));
return new IndexResult(plan.versionForIndexing, plan.seqNoForIndexing, plan.currentNotFoundOrDeleted);
} catch (Exception ex) {
if (indexWriter.getTragicException() == null) {
Expand All @@ -941,6 +963,13 @@ private IndexResult indexIntoLucene(Index index, IndexingStrategy plan)
}
}

private VersionValue getVersionValue(long version, long seqNo, long term, Translog.Location location) {
if (location != null && trackTranslogLocation.get()) {
return new TranslogVersionValue(location, version, seqNo, term);
}
return new VersionValue(version, seqNo, term);
}

/**
* returns true if the indexing operation may have already be processed by this engine.
* Note that it is OK to rarely return true even if this is not the case. However a `false`
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,207 @@
/*
* Licensed to Elasticsearch under one or more contributor
* license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright
* ownership. Elasticsearch 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.
*/
package org.elasticsearch.index.engine;

import org.apache.lucene.index.BinaryDocValues;
import org.apache.lucene.index.DocValuesType;
import org.apache.lucene.index.FieldInfo;
import org.apache.lucene.index.FieldInfos;
import org.apache.lucene.index.Fields;
import org.apache.lucene.index.IndexOptions;
import org.apache.lucene.index.LeafMetaData;
import org.apache.lucene.index.LeafReader;
import org.apache.lucene.index.NumericDocValues;
import org.apache.lucene.index.PointValues;
import org.apache.lucene.index.SortedDocValues;
import org.apache.lucene.index.SortedNumericDocValues;
import org.apache.lucene.index.SortedSetDocValues;
import org.apache.lucene.index.StoredFieldVisitor;
import org.apache.lucene.index.Terms;
import org.apache.lucene.util.Bits;
import org.apache.lucene.util.BytesRef;
import org.elasticsearch.index.fielddata.AbstractSortedDocValues;
import org.elasticsearch.index.fielddata.AbstractSortedSetDocValues;
import org.elasticsearch.index.mapper.ParentFieldMapper;
import org.elasticsearch.index.mapper.RoutingFieldMapper;
import org.elasticsearch.index.mapper.SourceFieldMapper;
import org.elasticsearch.index.translog.Translog;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Collections;

/**
* Internal class that mocks a single doc read from the transaction log as a leaf reader.
*/
final class TranslogLeafReader extends LeafReader {

private final Translog.Index operation;
private static final FieldInfo FAKE_SOURCE_FIELD
= new FieldInfo(SourceFieldMapper.NAME, 1, false, false, false, IndexOptions.NONE, DocValuesType.NONE, -1, Collections.emptyMap(),
0,0);
private static final FieldInfo FAKE_ROUTING_FIELD
= new FieldInfo(RoutingFieldMapper.NAME, 2, false, false, false, IndexOptions.NONE, DocValuesType.NONE, -1, Collections.emptyMap(),
0,0);

TranslogLeafReader(Translog.Index operation) {
this.operation = operation;
}
@Override
public CacheHelper getCoreCacheHelper() {
throw new UnsupportedOperationException();
}

@Override
public Terms terms(String field) {
throw new UnsupportedOperationException();
}

@Override
public NumericDocValues getNumericDocValues(String field) {
throw new UnsupportedOperationException();
}

@Override
public BinaryDocValues getBinaryDocValues(String field) {
throw new UnsupportedOperationException();
}

@Override
public SortedDocValues getSortedDocValues(String field) {
// TODO this can be removed in 7.0 and upwards we don't support the parent field anymore
if (field.startsWith(ParentFieldMapper.NAME + "#") && operation.parent() != null) {
return new AbstractSortedDocValues() {
@Override
public int docID() {
return 0;
}

private final BytesRef term = new BytesRef(operation.parent());
private int ord;
@Override
public boolean advanceExact(int docID) {
if (docID != 0) {
throw new IndexOutOfBoundsException("do such doc ID: " + docID);
}
ord = 0;
return true;
}

@Override
public int ordValue() {
return ord;
}

@Override
public BytesRef lookupOrd(int ord) {
if (ord == 0) {
return term;
}
return null;
}

@Override
public int getValueCount() {
return 1;
}
};
}
return null;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we throw an exception?

}

@Override
public SortedNumericDocValues getSortedNumericDocValues(String field) {
throw new UnsupportedOperationException();
}

@Override
public SortedSetDocValues getSortedSetDocValues(String field) {
throw new UnsupportedOperationException();
}

@Override
public NumericDocValues getNormValues(String field) {
throw new UnsupportedOperationException();
}

@Override
public FieldInfos getFieldInfos() {
throw new UnsupportedOperationException();
}

@Override
public Bits getLiveDocs() {
throw new UnsupportedOperationException();
}

@Override
public PointValues getPointValues(String field) {
throw new UnsupportedOperationException();
}

@Override
public void checkIntegrity() {

}

@Override
public LeafMetaData getMetaData() {
throw new UnsupportedOperationException();
}

@Override
public Fields getTermVectors(int docID) {
throw new UnsupportedOperationException();
}

@Override
public int numDocs() {
return 1;
}

@Override
public int maxDoc() {
return 1;
}

@Override
public void document(int docID, StoredFieldVisitor visitor) throws IOException {
if (docID != 0) {
throw new IllegalArgumentException("no such doc ID " + docID);
}
if (visitor.needsField(FAKE_SOURCE_FIELD) == StoredFieldVisitor.Status.YES) {
assert operation.source().toBytesRef().offset == 0;
assert operation.source().toBytesRef().length == operation.source().toBytesRef().bytes.length;
visitor.binaryField(FAKE_SOURCE_FIELD, operation.source().toBytesRef().bytes);
}
if (operation.routing() != null && visitor.needsField(FAKE_ROUTING_FIELD) == StoredFieldVisitor.Status.YES) {
visitor.stringField(FAKE_ROUTING_FIELD, operation.routing().getBytes(StandardCharsets.UTF_8));
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I know some visitors won't like that _id is not seen, so maybe assert that visitor.needsField("_id") is false?

}

@Override
protected void doClose() {

}

@Override
public CacheHelper getReaderCacheHelper() {
throw new UnsupportedOperationException();
}
}
Loading