Uses auto generated timestamp with soft-deletes

[dnhatn]

1. Peer-recovery

Today we don't store the auto-generated timestamp of indexing operations in Lucene and always assign -1 to all index operations from LuceneChangesSnapshot. This looks innocent but it generates duplicate documents on a replica in the following test.

public void testRetryAppendOnlyInRecoveryAndReplication() throws Exception {
    Settings settings = Settings.builder()
        .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true)
        .build();
    try (ReplicationGroup shards = createGroup(0, settings)) {
        shards.startAll();
        final IndexRequest originalRequest = new IndexRequest(
            index.getName(), "type").source("{}", XContentType.JSON);
        originalRequest.process(Version.CURRENT, null, index.getName());
        IndexRequest retryRequest = new IndexRequest();
        try (BytesStreamOutput out = new BytesStreamOutput()) {
            originalRequest.writeTo(out);
            try (StreamInput in = out.bytes().streamInput()) {
                retryRequest.readFrom(in);
            }
        }
        retryRequest.onRetry();
        shards.index(retryRequest);
        IndexShard replica = shards.addReplica();
        shards.recoverReplica(replica); // timestamp on replica is -1
        shards.assertAllEqual(1);
        shards.index(originalRequest); // we optimize this request on replica
        shards.assertAllEqual(1);
    }
}

To fix this, we need to assign a value which is at least the (original) timestamp of the index request to its corresponding index request from LucenChangeSnapshot. Here we can use the latest auto-generated timestamp of Engine.

2. Optimize indexing on a FollowingEngine in CCR

We disable optimization for index requests whose origin are recovery (retry always is true). To enable this optimization in CCR:

1. We need to make sure that a FollowingEngine processes an append-only operation once. This can be done using LocalCheckpointTracker.

2. We need to store the retry flag to Lucene index and extend Translog#Index to include this flag. This should be fast with a single value DocValues.

@s1monw WDYT? /cc @bleskes

This a subtask of #30086.

[elasticmachine]

Pinging @elastic/es-distributed

[s1monw]

To fix this, we need to assign a value which is at least the (original) timestamp of the index request to its corresponding index request from LucenChangeSnapshot. Here we can use the latest auto-generated timestamp of Engine.

I don't think we can do that. We don't know if the original request had such a timestamp or not. if we assign it to the wrong value we might miss an update. I think we should rather not try to optimize the peer recovery case and make sure we propagate the maxUnsafeAutoIdTimestamp from the primary to the replica before we replay the ops to it. that way we can be sure that if the retry / orig request arrives we deoptimize it and ops that come through peer recovery will be de-optimized.

We disable optimization for index requests whose origin are recovery (retry always is true). To enable this optimization in CCR:

I wonder if we can be smarter here. Today we can only optimize for append-only if the ID was autogenerated. Yet, for the following engine that's a different game. At the moment we fetch docs from the leader we could (if we had the information what was the highest seqID that actually did a delete or update in the engine) optimize also documents that have non-autogenerated IDs with append only. We can then use the seqID as the timestamp. Once we see the leader tells us that there is a seqID X that caused an update / delete we mark all ops with seqId <= X as retry. This will cause the right behavior for us. We also need to make sure that if we fetch docs a second time we mark them as retries. Lemme know if I am missing something.

[dnhatn]

I think we should rather not try to optimize the peer recovery case and make sure we propagate the maxUnsafeAutoIdTimestamp from the primary to the replica before we replay the ops to it.

Yes, I am good with this since we don't have to worry about the correctness.

At the moment we fetch docs from the leader we could (if we had the information what was the highest seqID that actually did a delete or update in the engine) optimize also documents that have non-autogenerated IDs with append only.

Wow, superb idea. Thanks so much @s1monw.

[dnhatn]

We will implement an optimization using sequence number on the following engine with these sub-tasks:

• Track max_seq_no_of_updates on the primary (Track max seq_no of updates or deletes on primary #33842)
• Replicate max_seq_no_updates to replicas in replication requests (Replicate max seq_no of updates to replicas #33967)
• Replicate max_seq_no_updates to follower in shard change requests (CCR: replicates max seq_no of updates to follower #34051)
• Apply optimization using max_seq_no_updates on the following engine (CCR: Optimize indexing ops using seq_no on followers #34099)

[dnhatn]

All subtasks have been integrated.