diff options
4 files changed, 58 insertions, 21 deletions
diff --git a/core/src/main/java/org/elasticsearch/index/engine/Engine.java b/core/src/main/java/org/elasticsearch/index/engine/Engine.java index 7763c8d04a..6e93d1feed 100644 --- a/core/src/main/java/org/elasticsearch/index/engine/Engine.java +++ b/core/src/main/java/org/elasticsearch/index/engine/Engine.java @@ -1102,8 +1102,8 @@ public abstract class Engine implements Closeable { public Delete(String type, String id, Term uid, long seqNo, long primaryTerm, long version, VersionType versionType, Origin origin, long startTime) { super(uid, seqNo, primaryTerm, version, versionType, origin, startTime); - this.type = type; - this.id = id; + this.type = Objects.requireNonNull(type); + this.id = Objects.requireNonNull(id); } public Delete(String type, String id, Term uid) { diff --git a/core/src/main/java/org/elasticsearch/index/translog/Translog.java b/core/src/main/java/org/elasticsearch/index/translog/Translog.java index 032d26d890..d4a5fe0d99 100644 --- a/core/src/main/java/org/elasticsearch/index/translog/Translog.java +++ b/core/src/main/java/org/elasticsearch/index/translog/Translog.java @@ -42,6 +42,7 @@ import org.elasticsearch.common.util.concurrent.ReleasableLock; import org.elasticsearch.index.IndexSettings; import org.elasticsearch.index.VersionType; import org.elasticsearch.index.engine.Engine; +import org.elasticsearch.index.mapper.Uid; import org.elasticsearch.index.seqno.SequenceNumbersService; import org.elasticsearch.index.shard.AbstractIndexShardComponent; import org.elasticsearch.index.shard.IndexShardComponent; @@ -58,6 +59,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.Iterator; import java.util.List; +import java.util.Objects; import java.util.Optional; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.ReadWriteLock; @@ -919,8 +921,8 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC private final String id; private final long autoGeneratedIdTimestamp; private final String type; - private long seqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO; - private long primaryTerm = 0; + private final long seqNo; + private final long primaryTerm; private final long version; private final VersionType versionType; private final BytesReference source; @@ -950,6 +952,9 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC if (format >= FORMAT_SEQ_NO) { seqNo = in.readLong(); primaryTerm = in.readLong(); + } else { + seqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO; + primaryTerm = 0; } } @@ -976,6 +981,7 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC this.id = id; this.source = new BytesArray(source); this.seqNo = seqNo; + this.primaryTerm = 0; this.version = version; this.versionType = versionType; this.routing = routing; @@ -1113,27 +1119,42 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC public static class Delete implements Operation { - private static final int FORMAT_5_X = 2; - private static final int FORMAT_SEQ_NO = FORMAT_5_X + 1; + public static final int FORMAT_5_0 = 2; // 5.0 - 5.5 + private static final int FORMAT_SINGLE_TYPE = FORMAT_5_0 + 1; // 5.5 - 6.0 + private static final int FORMAT_SEQ_NO = FORMAT_SINGLE_TYPE + 1; // 6.0 - * public static final int SERIALIZATION_FORMAT = FORMAT_SEQ_NO; - private String type, id; - private Term uid; - private long seqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO; - private long primaryTerm = 0; - private long version = Versions.MATCH_ANY; - private VersionType versionType = VersionType.INTERNAL; + private final String type, id; + private final Term uid; + private final long seqNo; + private final long primaryTerm; + private final long version; + private final VersionType versionType; public Delete(StreamInput in) throws IOException { final int format = in.readVInt();// SERIALIZATION_FORMAT - assert format >= FORMAT_5_X : "format was: " + format; - uid = new Term(in.readString(), in.readString()); + assert format >= FORMAT_5_0 : "format was: " + format; + if (format >= FORMAT_SINGLE_TYPE) { + type = in.readString(); + id = in.readString(); + uid = new Term(in.readString(), in.readString()); + } else { + uid = new Term(in.readString(), in.readString()); + // the uid was constructed from the type and id so we can + // extract them back + Uid uidObject = Uid.createUid(uid.text()); + type = uidObject.type(); + id = uidObject.id(); + } this.version = in.readLong(); this.versionType = VersionType.fromValue(in.readByte()); assert versionType.validateVersionForWrites(this.version); if (format >= FORMAT_SEQ_NO) { seqNo = in.readLong(); primaryTerm = in.readLong(); + } else { + seqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO; + primaryTerm = 0; } } @@ -1147,8 +1168,8 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC } public Delete(String type, String id, Term uid, long seqNo, long primaryTerm, long version, VersionType versionType) { - this.type = type; - this.id = id; + this.type = Objects.requireNonNull(type); + this.id = Objects.requireNonNull(id); this.uid = uid; this.seqNo = seqNo; this.primaryTerm = primaryTerm; @@ -1204,6 +1225,8 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC @Override public void writeTo(StreamOutput out) throws IOException { out.writeVInt(SERIALIZATION_FORMAT); + out.writeString(type); + out.writeString(id); out.writeString(uid.field()); out.writeString(uid.text()); out.writeLong(version); diff --git a/core/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java b/core/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java index 31a99063fb..16e746a67f 100644 --- a/core/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java +++ b/core/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java @@ -1939,7 +1939,7 @@ public class InternalEngineTests extends ESTestCase { indexResult = engine.index(index); assertFalse(indexResult.isCreated()); - engine.delete(new Engine.Delete(null, "1", newUid(doc))); + engine.delete(new Engine.Delete("doc", "1", newUid(doc))); index = indexForDoc(doc); indexResult = engine.index(index); diff --git a/core/src/test/java/org/elasticsearch/index/translog/TranslogTests.java b/core/src/test/java/org/elasticsearch/index/translog/TranslogTests.java index b911e9a5a4..ee7073fd53 100644 --- a/core/src/test/java/org/elasticsearch/index/translog/TranslogTests.java +++ b/core/src/test/java/org/elasticsearch/index/translog/TranslogTests.java @@ -354,24 +354,24 @@ public class TranslogTests extends ESTestCase { { final TranslogStats stats = stats(); assertThat(stats.estimatedNumberOfOperations(), equalTo(2L)); - assertThat(stats.getTranslogSizeInBytes(), equalTo(139L)); + assertThat(stats.getTranslogSizeInBytes(), equalTo(146L)); } translog.add(new Translog.Delete("test", "3", 2, newUid("3"))); { final TranslogStats stats = stats(); assertThat(stats.estimatedNumberOfOperations(), equalTo(3L)); - assertThat(stats.getTranslogSizeInBytes(), equalTo(181L)); + assertThat(stats.getTranslogSizeInBytes(), equalTo(195L)); } translog.add(new Translog.NoOp(3, 1, randomAlphaOfLength(16))); { final TranslogStats stats = stats(); assertThat(stats.estimatedNumberOfOperations(), equalTo(4L)); - assertThat(stats.getTranslogSizeInBytes(), equalTo(223L)); + assertThat(stats.getTranslogSizeInBytes(), equalTo(237L)); } - final long expectedSizeInBytes = 266L; + final long expectedSizeInBytes = 280L; translog.rollGeneration(); { final TranslogStats stats = stats(); @@ -2263,6 +2263,20 @@ public class TranslogTests extends ESTestCase { in = out.bytes().streamInput(); Translog.Delete serializedDelete = new Translog.Delete(in); assertEquals(delete, serializedDelete); + + // simulate legacy delete serialization + out = new BytesStreamOutput(); + out.writeVInt(Translog.Delete.FORMAT_5_0); + out.writeString(UidFieldMapper.NAME); + out.writeString("my_type#my_id"); + out.writeLong(3); // version + out.writeByte(VersionType.INTERNAL.getValue()); + out.writeLong(2); // seq no + out.writeLong(0); // primary term + in = out.bytes().streamInput(); + serializedDelete = new Translog.Delete(in); + assertEquals("my_type", serializedDelete.type()); + assertEquals("my_id", serializedDelete.id()); } public void testRollGeneration() throws IOException { |