[Transform] Reduce indexes to query based on checkpoints

[hendrikmuhs]

Continuous transform reduce the amount of data to query for by detecting what has been changed
since the last checkpoint. This information is than used to inject queries that narrow the
scope. The query is send to all configured indices. This change reduces the indexes to call
using checkpoint information. This reduces not only the number of network calls, but also
reduces the probability of a failure, which is more likely to happen in large heterogeneous
clusters (hot/warm/cold architecture).

[elasticmachine]

Pinging @elastic/ml-core (Team:ML)

[hendrikmuhs]

TL/DR

I think I need to explain the idea for part 2, reducing calls for pivot:

The change collector already collects meta information about the source indexes, e.g. it gets max(@timestamp). I want to keep this information and store it in the state document. Imagine we keep the last 5 checkpoints (whether its 5 or something higher needs to be tested):

_transform.cp.42.@timestamp.max: 999
_transform.cp.43.@timestamp.max: 1022
_transform.cp.44.@timestamp.max: 1080
_transform.cp.45.@timestamp.max: 1129
_transform.cp.46.@timestamp.max: 1535

Assume for checkpoint 47 we detected in the change collector that @timestamp has a range min: 1092, max: 1693. The above lookup table tells us, we can skip every index that hasn't changed after checkpoint 44. Therefore we call getChangedIndices(44, 47) which gives us the indices with changes.

In a nutshell the checkpoint for 44 could look like this:

index-0001: 1220,1220,10135,1015,10111
index-0002: 305,15134,13531,1352,15533
...
index-0099: 10, 10, 10, 10, 10
index-0101: 20, 30, 30, 30, 30
index-0102: 40, 40, 40, 40, 40

The checkpoint of 47 might be:

index-0001: 1220,1220,10135,1015,10111
index-0002: 305,15134,13531,1352,15533
...
index-0099: 10, 10, 10, 10, 10
index-0101: 23, 32, 42, 32, 60
index-0102: 40, 50, 70, 60, 50
index-0103: 10, 10, 10, 10, 10

If we diff the 2, we can remove [index-0001, ... index-0099], only these 3 indexes are of interest:

index-0101: 23, 32, 42, 32, 60
index-0102: 40, 50, 70, 60, 50
index-0103: 10, 10, 10, 10, 10

We therefore can resolve the pattern index-* to only call the shards of those 3 indexes instead of calling all of them.

[benwtrent]

I think the idea is sound and will definitely help resiliency. I don't think it will help much in the performance front.

I would like to add the name for the PIT and searchrequest to the logging messages in some fashion.

Additionally, I am slightly concerned that building the checkpoints themselves will still fail as they have to get the stats from all indices. Though, I am not 100% sure how that index stats request differs from searching them.

[benwtrent]

I really like this. Naming them like this makes debugging in the future much nicer!!!

[hendrikmuhs]

@elasticmachine update branch

[hendrikmuhs]

Additionally, I am slightly concerned that building the checkpoints themselves will still fail as they have to get the stats from all indices. Though, I am not 100% sure how that index stats request differs from searching them.

For the record / others reading this issue: This is tracked and followed up in #75780.

An index stats call still requires a network call, so we won't get away with this. However index stats is answered from the in-memory state, a search - whether it returns a match or not - potentially causes disk IO. The index stats is done exactly once per checkpoint, search gets executed several times.

It's all baby steps and involves a lot "it depends".

[elasticsearchmachine]

💚 Backport successful

Status	Branch	Result
✅	7.x