Skip to content

Commit d621fc7

Browse files
dnhatndnhatn
authored
Add tombstone document into Lucene for Noop (#30226)
This commit adds a tombstone document into Lucene for every No-op. With this change, Lucene index is expected to have a complete history of operations like Translog. In fact, this guarantee is subjected to the soft-deletes retention merge policy. Relates #29530
1 parent 217d090 commit d621fc7

File tree

17 files changed

+328
-66
lines changed

17 files changed

+328
-66
lines changed

‎server/src/main/java/org/elasticsearch/index/engine/EngineConfig.java

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -372,9 +372,16 @@ public LongSupplier getPrimaryTermSupplier() {
372372
* A supplier supplies tombstone documents which will be used in soft-update methods.
373373
* The returned document consists only _uid, _seqno, _term and _version fields; other metadata fields are excluded.
374374
*/
375-
@FunctionalInterface
376375
public interface TombstoneDocSupplier {
377-
ParsedDocument newTombstoneDoc(String type, String id);
376+
/**
377+
* Creates a tombstone document for a delete operation.
378+
*/
379+
ParsedDocument newDeleteTombstoneDoc(String type, String id);
380+
381+
/**
382+
* Creates a tombstone document for a noop operation.
383+
*/
384+
ParsedDocument newNoopTombstoneDoc();
378385
}
379386

380387
public TombstoneDocSupplier getTombstoneDocSupplier() {

‎server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

Lines changed: 26 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@
6767
import org.elasticsearch.index.mapper.IdFieldMapper;
6868
import org.elasticsearch.index.mapper.ParseContext;
6969
import org.elasticsearch.index.mapper.ParsedDocument;
70+
import org.elasticsearch.index.mapper.SeqNoFieldMapper;
7071
import org.elasticsearch.index.merge.MergeStats;
7172
import org.elasticsearch.index.merge.OnGoingMerge;
7273
import org.elasticsearch.index.seqno.LocalCheckpointTracker;
@@ -784,7 +785,9 @@ public IndexResult index(Index index) throws IOException {
784785
location = translog.add(new Translog.Index(index, indexResult));
785786
} else if (indexResult.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
786787
// if we have document failure, record it as a no-op in the translog with the generated seq_no
787-
location = translog.add(new Translog.NoOp(indexResult.getSeqNo(), index.primaryTerm(), indexResult.getFailure().getMessage()));
788+
final NoOp noOp = new NoOp(indexResult.getSeqNo(), index.primaryTerm(), index.origin(),
789+
index.startTime(), indexResult.getFailure().getMessage());
790+
location = innerNoOp(noOp).getTranslogLocation();
788791
} else {
789792
location = null;
790793
}
@@ -1226,11 +1229,13 @@ private DeleteResult deleteInLucene(Delete delete, DeletionStrategy plan)
12261229
throws IOException {
12271230
try {
12281231
if (softDeleteEnabled) {
1229-
final ParsedDocument tombstone = engineConfig.getTombstoneDocSupplier().newTombstoneDoc(delete.type(), delete.id());
1232+
final ParsedDocument tombstone = engineConfig.getTombstoneDocSupplier().newDeleteTombstoneDoc(delete.type(), delete.id());
12301233
assert tombstone.docs().size() == 1 : "Tombstone doc should have single doc [" + tombstone + "]";
12311234
tombstone.updateSeqID(plan.seqNoOfDeletion, delete.primaryTerm());
12321235
tombstone.version().setLongValue(plan.versionOfDeletion);
12331236
final ParseContext.Document doc = tombstone.docs().get(0);
1237+
assert doc.getField(SeqNoFieldMapper.TOMBSTONE_NAME) != null :
1238+
"Delete tombstone document but _tombstone field is not set [" + doc + " ]";
12341239
doc.add(softDeleteField);
12351240
if (plan.addStaleOpToLucene || plan.currentlyDeleted) {
12361241
indexWriter.addDocument(doc);
@@ -1334,7 +1339,25 @@ private NoOpResult innerNoOp(final NoOp noOp) throws IOException {
13341339
assert noOp.seqNo() > SequenceNumbers.NO_OPS_PERFORMED;
13351340
final long seqNo = noOp.seqNo();
13361341
try {
1337-
final NoOpResult noOpResult = new NoOpResult(noOp.seqNo());
1342+
Exception failure = null;
1343+
if (softDeleteEnabled) {
1344+
try {
1345+
final ParsedDocument tombstone = engineConfig.getTombstoneDocSupplier().newNoopTombstoneDoc();
1346+
tombstone.updateSeqID(noOp.seqNo(), noOp.primaryTerm());
1347+
assert tombstone.docs().size() == 1 : "Tombstone should have a single doc [" + tombstone + "]";
1348+
final ParseContext.Document doc = tombstone.docs().get(0);
1349+
assert doc.getField(SeqNoFieldMapper.TOMBSTONE_NAME) != null
1350+
: "Noop tombstone document but _tombstone field is not set [" + doc + " ]";
1351+
doc.add(softDeleteField);
1352+
indexWriter.addDocument(doc);
1353+
} catch (Exception ex) {
1354+
if (maybeFailEngine("noop", ex)) {
1355+
throw ex;
1356+
}
1357+
failure = ex;
1358+
}
1359+
}
1360+
final NoOpResult noOpResult = failure != null ? new NoOpResult(noOp.seqNo(), failure) : new NoOpResult(noOp.seqNo());
13381361
if (noOp.origin() != Operation.Origin.LOCAL_TRANSLOG_RECOVERY) {
13391362
final Translog.Location location = translog.add(new Translog.NoOp(noOp.seqNo(), noOp.primaryTerm(), noOp.reason()));
13401363
noOpResult.setTranslogLocation(location);

‎server/src/main/java/org/elasticsearch/index/mapper/DocumentMapper.java

Lines changed: 19 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -124,7 +124,8 @@ public DocumentMapper build(MapperService mapperService) {
124124
private final Map<String, ObjectMapper> objectMappers;
125125

126126
private final boolean hasNestedObjects;
127-
private final MetadataFieldMapper[] tombstoneMetadataFieldMappers;
127+
private final MetadataFieldMapper[] deleteTombstoneMetadataFieldMappers;
128+
private final MetadataFieldMapper[] noopTombstoneMetadataFieldMappers;
128129

129130
public DocumentMapper(MapperService mapperService, Mapping mapping) {
130131
this.mapperService = mapperService;
@@ -133,10 +134,6 @@ public DocumentMapper(MapperService mapperService, Mapping mapping) {
133134
final IndexSettings indexSettings = mapperService.getIndexSettings();
134135
this.mapping = mapping;
135136
this.documentParser = new DocumentParser(indexSettings, mapperService.documentMapperParser(), this);
136-
final Collection<String> tombstoneFields =
137-
Arrays.asList(SeqNoFieldMapper.NAME, SeqNoFieldMapper.PRIMARY_TERM_NAME, VersionFieldMapper.NAME, IdFieldMapper.NAME);
138-
this.tombstoneMetadataFieldMappers = Stream.of(mapping.metadataMappers)
139-
.filter(field -> tombstoneFields.contains(field.name())).toArray(MetadataFieldMapper[]::new);
140137

141138
// collect all the mappers for this type
142139
List<ObjectMapper> newObjectMappers = new ArrayList<>();
@@ -176,6 +173,15 @@ public DocumentMapper(MapperService mapperService, Mapping mapping) {
176173
} catch (Exception e) {
177174
throw new ElasticsearchGenerationException("failed to serialize source for type [" + type + "]", e);
178175
}
176+
177+
final Collection<String> deleteTombstoneMetadataFields = Arrays.asList(VersionFieldMapper.NAME, IdFieldMapper.NAME,
178+
TypeFieldMapper.NAME, SeqNoFieldMapper.NAME, SeqNoFieldMapper.PRIMARY_TERM_NAME, SeqNoFieldMapper.TOMBSTONE_NAME);
179+
this.deleteTombstoneMetadataFieldMappers = Stream.of(mapping.metadataMappers)
180+
.filter(field -> deleteTombstoneMetadataFields.contains(field.name())).toArray(MetadataFieldMapper[]::new);
181+
final Collection<String> noopTombstoneMetadataFields =
182+
Arrays.asList(SeqNoFieldMapper.NAME, SeqNoFieldMapper.PRIMARY_TERM_NAME, SeqNoFieldMapper.TOMBSTONE_NAME);
183+
this.noopTombstoneMetadataFieldMappers = Stream.of(mapping.metadataMappers)
184+
.filter(field -> noopTombstoneMetadataFields.contains(field.name())).toArray(MetadataFieldMapper[]::new);
179185
}
180186

181187
public Mapping mapping() {
@@ -251,9 +257,15 @@ public ParsedDocument parse(SourceToParse source) throws MapperParsingException
251257
return documentParser.parseDocument(source, mapping.metadataMappers);
252258
}
253259

254-
public ParsedDocument createTombstoneDoc(String index, String type, String id) throws MapperParsingException {
260+
public ParsedDocument createDeleteTombstoneDoc(String index, String type, String id) throws MapperParsingException {
261+
final SourceToParse emptySource = SourceToParse.source(index, type, id, new BytesArray("{}"), XContentType.JSON);
262+
return documentParser.parseDocument(emptySource, deleteTombstoneMetadataFieldMappers).toTombstone();
263+
}
264+
265+
public ParsedDocument createNoopTombstoneDoc(String index) throws MapperParsingException {
266+
final String id = ""; // _id won't be used.
255267
final SourceToParse emptySource = SourceToParse.source(index, type, id, new BytesArray("{}"), XContentType.JSON);
256-
return documentParser.parseDocument(emptySource, tombstoneMetadataFieldMappers);
268+
return documentParser.parseDocument(emptySource, noopTombstoneMetadataFieldMappers).toTombstone();
257269
}
258270

259271
/**

‎server/src/main/java/org/elasticsearch/index/mapper/ParsedDocument.java

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,17 @@ public void updateSeqID(long sequenceNumber, long primaryTerm) {
8383
this.seqID.primaryTerm.setLongValue(primaryTerm);
8484
}
8585

86+
/**
87+
* Makes the processing document as a tombstone document rather than a regular document.
88+
* Tombstone documents are stored in Lucene index to represent delete operations or Noops.
89+
*/
90+
ParsedDocument toTombstone() {
91+
assert docs().size() == 1 : "Tombstone should have a single doc [" + docs() + "]";
92+
this.seqID.tombstoneField.setLongValue(1);
93+
rootDoc().add(this.seqID.tombstoneField);
94+
return this;
95+
}
96+
8697
public String routing() {
8798
return this.routing;
8899
}

‎server/src/main/java/org/elasticsearch/index/mapper/SeqNoFieldMapper.java

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -69,26 +69,29 @@ public static class SequenceIDFields {
6969
public final Field seqNo;
7070
public final Field seqNoDocValue;
7171
public final Field primaryTerm;
72+
public final Field tombstoneField;
7273

73-
public SequenceIDFields(Field seqNo, Field seqNoDocValue, Field primaryTerm) {
74+
public SequenceIDFields(Field seqNo, Field seqNoDocValue, Field primaryTerm, Field tombstoneField) {
7475
Objects.requireNonNull(seqNo, "sequence number field cannot be null");
7576
Objects.requireNonNull(seqNoDocValue, "sequence number dv field cannot be null");
7677
Objects.requireNonNull(primaryTerm, "primary term field cannot be null");
7778
this.seqNo = seqNo;
7879
this.seqNoDocValue = seqNoDocValue;
7980
this.primaryTerm = primaryTerm;
81+
this.tombstoneField = tombstoneField;
8082
}
8183

8284
public static SequenceIDFields emptySeqID() {
8385
return new SequenceIDFields(new LongPoint(NAME, SequenceNumbers.UNASSIGNED_SEQ_NO),
8486
new NumericDocValuesField(NAME, SequenceNumbers.UNASSIGNED_SEQ_NO),
85-
new NumericDocValuesField(PRIMARY_TERM_NAME, 0));
87+
new NumericDocValuesField(PRIMARY_TERM_NAME, 0), new NumericDocValuesField(TOMBSTONE_NAME, 0));
8688
}
8789
}
8890

8991
public static final String NAME = "_seq_no";
9092
public static final String CONTENT_TYPE = "_seq_no";
9193
public static final String PRIMARY_TERM_NAME = "_primary_term";
94+
public static final String TOMBSTONE_NAME = "_tombstone";
9295

9396
public static class SeqNoDefaults {
9497
public static final String NAME = SeqNoFieldMapper.NAME;

‎server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

Lines changed: 16 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -90,12 +90,14 @@
9090
import org.elasticsearch.index.flush.FlushStats;
9191
import org.elasticsearch.index.get.GetStats;
9292
import org.elasticsearch.index.get.ShardGetService;
93+
import org.elasticsearch.index.mapper.DocumentMapper;
9394
import org.elasticsearch.index.mapper.DocumentMapperForType;
9495
import org.elasticsearch.index.mapper.IdFieldMapper;
9596
import org.elasticsearch.index.mapper.MapperParsingException;
9697
import org.elasticsearch.index.mapper.MapperService;
9798
import org.elasticsearch.index.mapper.Mapping;
9899
import org.elasticsearch.index.mapper.ParsedDocument;
100+
import org.elasticsearch.index.mapper.RootObjectMapper;
99101
import org.elasticsearch.index.mapper.SourceToParse;
100102
import org.elasticsearch.index.mapper.Uid;
101103
import org.elasticsearch.index.merge.MergeStats;
@@ -2162,8 +2164,7 @@ private EngineConfig newEngineConfig() {
21622164
IndexingMemoryController.SHARD_INACTIVE_TIME_SETTING.get(indexSettings.getSettings()),
21632165
Collections.singletonList(refreshListeners),
21642166
Collections.singletonList(new RefreshMetricUpdater(refreshMetric)),
2165-
indexSort, this::runTranslogRecovery, circuitBreakerService, replicationTracker, this::getPrimaryTerm,
2166-
this::createTombstoneDoc);
2167+
indexSort, this::runTranslogRecovery, circuitBreakerService, replicationTracker, this::getPrimaryTerm, tombstoneDocSupplier());
21672168
}
21682169

21692170
/**
@@ -2592,7 +2593,18 @@ public void afterRefresh(boolean didRefresh) throws IOException {
25922593
}
25932594
}
25942595

2595-
private ParsedDocument createTombstoneDoc(String type, String id) {
2596-
return docMapper(type).getDocumentMapper().createTombstoneDoc(shardId.getIndexName(), type, id);
2596+
private EngineConfig.TombstoneDocSupplier tombstoneDocSupplier() {
2597+
final RootObjectMapper.Builder noopRootMapper = new RootObjectMapper.Builder("__noop");
2598+
final DocumentMapper noopDocumentMapper = new DocumentMapper.Builder(noopRootMapper, mapperService).build(mapperService);
2599+
return new EngineConfig.TombstoneDocSupplier() {
2600+
@Override
2601+
public ParsedDocument newDeleteTombstoneDoc(String type, String id) {
2602+
return docMapper(type).getDocumentMapper().createDeleteTombstoneDoc(shardId.getIndexName(), type, id);
2603+
}
2604+
@Override
2605+
public ParsedDocument newNoopTombstoneDoc() {
2606+
return noopDocumentMapper.createNoopTombstoneDoc(shardId.getIndexName());
2607+
}
2608+
};
25972609
}
25982610
}

‎server/src/test/java/org/elasticsearch/cluster/routing/PrimaryAllocationIT.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -391,6 +391,7 @@ public void testPrimaryReplicaResyncFailed() throws Exception {
391391
assertThat(shard.getLocalCheckpoint(), equalTo(numDocs + moreDocs));
392392
}
393393
}, 30, TimeUnit.SECONDS);
394+
internalCluster().assertConsistentHistoryBetweenTranslogAndLuceneIndex();
394395
}
395396

396397
}

‎server/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java

Lines changed: 19 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,7 @@
103103
import org.elasticsearch.index.mapper.ContentPath;
104104
import org.elasticsearch.index.mapper.IdFieldMapper;
105105
import org.elasticsearch.index.mapper.Mapper.BuilderContext;
106+
import org.elasticsearch.index.mapper.MapperService;
106107
import org.elasticsearch.index.mapper.Mapping;
107108
import org.elasticsearch.index.mapper.MetadataFieldMapper;
108109
import org.elasticsearch.index.mapper.ParseContext;
@@ -173,6 +174,7 @@
173174
import static org.hamcrest.Matchers.everyItem;
174175
import static org.hamcrest.Matchers.greaterThan;
175176
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
177+
import static org.hamcrest.Matchers.hasItem;
176178
import static org.hamcrest.Matchers.hasKey;
177179
import static org.hamcrest.Matchers.hasSize;
178180
import static org.hamcrest.Matchers.lessThanOrEqualTo;
@@ -2603,7 +2605,7 @@ public void testRecoverFromForeignTranslog() throws IOException {
26032605
new CodecService(null, logger), config.getEventListener(), IndexSearcher.getDefaultQueryCache(),
26042606
IndexSearcher.getDefaultQueryCachingPolicy(), translogConfig, TimeValue.timeValueMinutes(5),
26052607
config.getExternalRefreshListener(), config.getInternalRefreshListener(), null, config.getTranslogRecoveryRunner(),
2606-
new NoneCircuitBreakerService(), () -> SequenceNumbers.UNASSIGNED_SEQ_NO, primaryTerm::get, EngineTestCase::createTombstoneDoc);
2608+
new NoneCircuitBreakerService(), () -> SequenceNumbers.UNASSIGNED_SEQ_NO, primaryTerm::get, tombstoneDocSupplier());
26072609
try {
26082610
InternalEngine internalEngine = new InternalEngine(brokenConfig);
26092611
fail("translog belongs to a different engine");
@@ -3046,7 +3048,8 @@ public void testDoubleDeliveryReplica() throws IOException {
30463048
TopDocs topDocs = searcher.searcher().search(new MatchAllDocsQuery(), 10);
30473049
assertEquals(1, topDocs.totalHits);
30483050
}
3049-
assertThat(getOperationSeqNoInLucene(engine), contains(20L));
3051+
List<Translog.Operation> ops = readAllOperationsInLucene(engine, createMapperService("test"));
3052+
assertThat(ops.stream().map(o -> o.seqNo()).collect(Collectors.toList()), hasItem(20L));
30503053
}
30513054

30523055
public void testRetryWithAutogeneratedIdWorksAndNoDuplicateDocs() throws IOException {
@@ -3606,7 +3609,9 @@ public void testNoOps() throws IOException {
36063609
maxSeqNo,
36073610
localCheckpoint);
36083611
trimUnsafeCommits(engine.config());
3609-
noOpEngine = new InternalEngine(engine.config(), supplier) {
3612+
EngineConfig noopEngineConfig = copy(engine.config(), new SoftDeletesRetentionMergePolicy(Lucene.SOFT_DELETE_FIELD,
3613+
() -> new MatchAllDocsQuery(), engine.config().getMergePolicy()));
3614+
noOpEngine = new InternalEngine(noopEngineConfig, supplier) {
36103615
@Override
36113616
protected long doGenerateSeqNoForOperation(Operation operation) {
36123617
throw new UnsupportedOperationException();
@@ -3636,6 +3641,13 @@ protected long doGenerateSeqNoForOperation(Operation operation) {
36363641
assertThat(noOp.seqNo(), equalTo((long) (maxSeqNo + 2)));
36373642
assertThat(noOp.primaryTerm(), equalTo(primaryTerm.get()));
36383643
assertThat(noOp.reason(), equalTo(reason));
3644+
MapperService mapperService = createMapperService("test");
3645+
List<Translog.Operation> operationsFromLucene = readAllOperationsInLucene(noOpEngine, mapperService);
3646+
assertThat(operationsFromLucene, hasSize(maxSeqNo + 2 - localCheckpoint)); // fills n gap and 2 manual noop.
3647+
for (int i = 0; i < operationsFromLucene.size(); i++) {
3648+
assertThat(operationsFromLucene.get(i), equalTo(new Translog.NoOp(localCheckpoint + 1 + i, primaryTerm.get(), "")));
3649+
}
3650+
assertConsistentHistoryBetweenTranslogAndLuceneIndex(noOpEngine, mapperService);
36393651
} finally {
36403652
IOUtils.close(noOpEngine);
36413653
}
@@ -4603,7 +4615,10 @@ private void assertOperationHistoryInLucene(List<Engine.Operation> operations) t
46034615
engine.forceMerge(true);
46044616
}
46054617
}
4606-
assertThat(getOperationSeqNoInLucene(engine), containsInAnyOrder(expectedSeqNos.toArray()));
4618+
MapperService mapperService = createMapperService("test");
4619+
List<Translog.Operation> actualOps = readAllOperationsInLucene(engine, mapperService);
4620+
assertThat(actualOps.stream().map(o -> o.seqNo()).collect(Collectors.toList()), containsInAnyOrder(expectedSeqNos.toArray()));
4621+
assertConsistentHistoryBetweenTranslogAndLuceneIndex(engine, mapperService);
46074622
}
46084623
}
46094624

‎server/src/test/java/org/elasticsearch/index/replication/IndexLevelReplicationTests.java

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -245,7 +245,11 @@ public void testDocumentFailureReplication() throws Exception {
245245
try (ReplicationGroup shards = new ReplicationGroup(buildIndexMetaData(0)) {
246246
@Override
247247
protected EngineFactory getEngineFactory(ShardRouting routing) {
248-
return throwingDocumentFailureEngineFactory;
248+
if (routing.primary()){
249+
return throwingDocumentFailureEngineFactory; // Simulate exception only on the primary.
250+
}else {
251+
return InternalEngine::new;
252+
}
249253
}}) {
250254

251255
// test only primary

0 commit comments

Comments
 (0)