Skip to content

Add internal _primary_term field to store shard's primary term #21480

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

Closed
wants to merge 1 commit into from
Closed

Add internal _primary_term field to store shard's primary term #21480

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
3 changes: 2 additions & 1 deletion core/src/main/java/org/elasticsearch/action/delete/TransportDeleteAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,7 +142,8 @@ public static WriteResult<DeleteResponse> executeDeleteRequestOnPrimary(DeleteRe
}

public static Engine.Delete executeDeleteRequestOnReplica(DeleteRequest request, IndexShard indexShard) {
Engine.Delete delete = indexShard.prepareDeleteOnReplica(request.type(), request.id(), request.seqNo(), request.version(), request.versionType());
Engine.Delete delete = indexShard.prepareDeleteOnReplica(request.type(), request.id(), request.seqNo(), request.primaryTerm(),
request.version(), request.versionType());
indexShard.delete(delete);
return delete;
}
Expand Down
28 changes: 19 additions & 9 deletions core/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 @@ -776,17 +776,20 @@ public abstract static class Operation {
private final Term uid;
private long version;
private long seqNo;
private long primaryTerm;
private final VersionType versionType;
private final Origin origin;
private Translog.Location location;
private final long startTime;
private long endTime;

public Operation(Term uid, long seqNo, long version, VersionType versionType, Origin origin, long startTime) {
public Operation(Term uid, long seqNo, long primaryTerm, long version, VersionType versionType, Origin origin, long startTime) {
this.uid = uid;
assert origin != Origin.PRIMARY || seqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO : "seqNo should not be set when origin is PRIMARY";
assert origin != Origin.PRIMARY || seqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO
: "seqNo should not be set when origin is PRIMARY";
assert origin == Origin.PRIMARY || seqNo >= 0 : "seqNo should be set when origin is not PRIMARY";
this.seqNo = seqNo;
this.primaryTerm = primaryTerm;
this.version = version;
this.versionType = versionType;
this.origin = origin;
Expand Down Expand Up @@ -828,6 +831,10 @@ public void updateSeqNo(long seqNo) {
this.seqNo = seqNo;
}

public long primaryTerm() {
return primaryTerm;
}

public void setTranslogLocation(Translog.Location location) {
this.location = location;
}
Expand Down Expand Up @@ -880,9 +887,9 @@ public static class Index extends Operation {
private final boolean isRetry;
private boolean created;

public Index(Term uid, ParsedDocument doc, long seqNo, long version, VersionType versionType, Origin origin, long startTime,
long autoGeneratedIdTimestamp, boolean isRetry) {
super(uid, seqNo, version, versionType, origin, startTime);
public Index(Term uid, ParsedDocument doc, long seqNo, long primaryTerm, long version, VersionType versionType, Origin origin,
long startTime, long autoGeneratedIdTimestamp, boolean isRetry) {
super(uid, seqNo, primaryTerm, version, versionType, origin, startTime);
this.doc = doc;
this.isRetry = isRetry;
this.autoGeneratedIdTimestamp = autoGeneratedIdTimestamp;
Expand All @@ -893,7 +900,8 @@ public Index(Term uid, ParsedDocument doc) {
} // TEST ONLY

Index(Term uid, ParsedDocument doc, long version) {
this(uid, doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, version, VersionType.INTERNAL, Origin.PRIMARY, System.nanoTime(), -1, false);
this(uid, doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, version, VersionType.INTERNAL,
Origin.PRIMARY, System.nanoTime(), -1, false);
}

public ParsedDocument parsedDoc() {
Expand Down Expand Up @@ -984,15 +992,17 @@ public static class Delete extends Operation {
private final String id;
private boolean found;

public Delete(String type, String id, Term uid, long seqNo, long version, VersionType versionType, Origin origin, long startTime, boolean found) {
super(uid, seqNo, version, versionType, origin, startTime);
public Delete(String type, String id, Term uid, long seqNo, long primaryTerm, long version, VersionType versionType,
Origin origin, long startTime, boolean found) {
super(uid, seqNo, primaryTerm, version, versionType, origin, startTime);
this.type = type;
this.id = id;
this.found = found;
}

public Delete(String type, String id, Term uid) {
this(type, id, uid, SequenceNumbersService.UNASSIGNED_SEQ_NO, Versions.MATCH_ANY, VersionType.INTERNAL, Origin.PRIMARY, System.nanoTime(), false);
this(type, id, uid, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_ANY, VersionType.INTERNAL,
Origin.PRIMARY, System.nanoTime(), false);
}

@Override
Expand Down
1 change: 1 addition & 0 deletions core/src/main/java/org/elasticsearch/index/mapper/DocumentParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -150,6 +150,7 @@ private static ParsedDocument parsedDocument(SourceToParse source, ParseContext.
return new ParsedDocument(
context.version(),
context.seqNo(),
context.primaryTerm(),
context.sourceToParse().id(),
context.sourceToParse().type(),
source.routing(),
Expand Down
25 changes: 25 additions & 0 deletions core/src/main/java/org/elasticsearch/index/mapper/ParseContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -263,6 +263,16 @@ public void seqNo(Field seqNo) {
in.seqNo(seqNo);
}

@Override
public Field primaryTerm() {
return in.primaryTerm();
}

@Override
public void primaryTerm(Field primaryTerm) {
in.primaryTerm(primaryTerm);
}

@Override
public AllEntries allEntries() {
return in.allEntries();
Expand Down Expand Up @@ -312,6 +322,8 @@ public static class InternalParseContext extends ParseContext {

private Field seqNo;

private Field primaryTerm;

private final AllEntries allEntries;

private final List<Mapper> dynamicMappers;
Expand Down Expand Up @@ -413,6 +425,15 @@ public void seqNo(Field seqNo) {
this.seqNo = seqNo;
}

@Override
public Field primaryTerm() {
return this.primaryTerm;
}

@Override
public void primaryTerm(Field primaryTerm) {
this.primaryTerm = primaryTerm;
}

@Override
public AllEntries allEntries() {
Expand Down Expand Up @@ -544,6 +565,10 @@ public boolean isWithinMultiFields() {

public abstract void seqNo(Field seqNo);

public abstract Field primaryTerm();

public abstract void primaryTerm(Field primaryTerm);

public final boolean includeInAll(Boolean includeInAll, FieldMapper mapper) {
return includeInAll(includeInAll, mapper.fieldType().indexOptions() != IndexOptions.NONE);
}
Expand Down
28 changes: 17 additions & 11 deletions core/src/main/java/org/elasticsearch/index/mapper/ParsedDocument.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ public class ParsedDocument {
private final String id, type;
private final BytesRef uid;
private final Field seqNo;
private final Field primaryTerm;

private final String routing;

Expand All @@ -51,19 +52,20 @@ public class ParsedDocument {

private String parent;

public ParsedDocument(
Field version,
Field seqNo,
String id,
String type,
String routing,
long timestamp,
long ttl,
List<Document> documents,
BytesReference source,
Mapping dynamicMappingsUpdate) {
public ParsedDocument(Field version,
Field seqNo,
Field primaryTerm,
String id,
String type,
String routing,
long timestamp,
long ttl,
List<Document> documents,
BytesReference source,
Mapping dynamicMappingsUpdate) {
this.version = version;
this.seqNo = seqNo;
this.primaryTerm = primaryTerm;
this.id = id;
this.type = type;
this.uid = Uid.createUidAsBytes(type, id);
Expand Down Expand Up @@ -95,6 +97,10 @@ public Field seqNo() {
return seqNo;
}

public Field primaryTerm() {
return primaryTerm;
}

public String routing() {
return this.routing;
}
Expand Down
Loading