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.

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

Jump to bottom

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

Merged
merged 5 commits into from
Mar 28, 2018
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a8d3b73
Allow _update and upsert to read from the transaction log
s1monw Mar 27, 2018
3fa0fc0
apply feedback from @jpountz
s1monw Mar 27, 2018
41a3c32
Merge branch 'master' into read_from_translog_for_updates
s1monw Mar 27, 2018
3b84354
apply feedback from @ywelch
s1monw Mar 28, 2018
5be76c5
add annontation
s1monw Mar 28, 2018
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: 2 additions & 2 deletions server/src/main/java/org/elasticsearch/action/explain/TransportExplainAction.java
View file
Open in desktop
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
6 changes: 2 additions & 4 deletions server/src/main/java/org/elasticsearch/action/update/UpdateHelper.java
View file
Open in desktop
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
2 changes: 1 addition & 1 deletion ...r/src/main/java/org/elasticsearch/common/lucene/uid/PerThreadIDVersionAndSeqNoLookup.java
View file
Open in desktop
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
9 changes: 6 additions & 3 deletions server/src/main/java/org/elasticsearch/common/lucene/uid/VersionsAndSeqNoResolver.java
View file
Open in desktop
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
8 changes: 7 additions & 1 deletion server/src/main/java/org/elasticsearch/index/engine/Engine.java
View file
Open in desktop
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
36 changes: 34 additions & 2 deletions server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java
View file
Open in desktop
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,27 @@ 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

// in the case of a already pruned translog generation we might get null here - yet very unlikely
TranslogLeafReader reader = new TranslogLeafReader((Translog.Index) operation, engineConfig
.getIndexSettings().getIndexVersionCreated());
return new GetResult(new Searcher("realtime_get", new IndexSearcher(reader)),
new VersionsAndSeqNoResolver.DocIdAndVersion(0, ((Translog.Index) operation).version(), reader, 0));
}
} catch (IOException e) {
maybeFailEngine("realtime_get", e); // lets check if the translog has failed with a tragic event
Copy link
Contributor

Choose a reason for hiding this comment

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

In other places, we have wrapped the maybeFailEngine call as follows:

try {
    maybeFailEngine("index", e);
} catch (Exception inner) {
    e.addSuppressed(inner);
}
throw ...;

Copy link
Contributor Author

Choose a reason for hiding this comment

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

that's bogus I think. There is no exception throw from maybeFaileEngine only errors which should not be handled. I think we are good here?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

I will clean this up in a followup

Copy link
Contributor

Choose a reason for hiding this comment

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

right, makes sense

throw new EngineException(shardId, "failed to read operation from translog", e);
}
} else {
trackTranslogLocation.set(true);
}
}
refresh("realtime_get", SearcherScope.INTERNAL);
}
scope = SearcherScope.INTERNAL;
Expand Down Expand Up @@ -790,6 +813,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 +943,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 +966,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
Loading