Hash the tsid to overcome dimensions limits

[salvatore-campagna]

A Lucene limitation on doc values for UTF-8 fields does not allow
us to write keyword fields whose size is larger then 32K. This limits
our ability to map more than a certain number of dimension fields
for time series indices. Before introducing this change the tsid is
created as a catenation of dimension field names and values into
a keyword field.

To overcome this limitation we hash the tsid. This PR is intended
to be used as a draft to test different options.

Note that, as a side effect, this reduces the size of the tsid field
as a result of storing far less data when the tsid is hashed. Anyway,
we expect tsid hashing to affect compression of doc values and
resulting in larger storage footprint. Effect on query latency
needs to be evaluated too.

Resolves #93564

[salvatore-campagna]

This breaks for sure downsampling persistent task assumption about _tsid ordering.
See #97557.

If we decide to proceed with this tsid hashing solution we can address the issue in the
downsampling PR by searching if the source index includes hashed tsids and eventually
starting from scratch that specific downsampling task. The idea is that we use tsid
hashing rarely and, as a result, we would incur this performance penalty rarely too.

[salvatore-campagna]

I need to comment this assertion, at least for now, because when we hash the tsid we do not serialise dimensions and routing fields which later on results in a failure since createId will throw an exception due to missing routing fields.

[martijnvg]

Maybe this is a matter of where we do the hashing? Maybe we should alter TimeSeriesIdBuilder class.
Just hash the timeSeriesId in TimeSeriesIdFieldMapper#postParse(...) after line 148 only if it is beyond the 32KB limit and store that as SortedDocValuesField. And just pass down the original id to TsidExtractingIdFieldMapper#createField(...)?

[martijnvg]

Can this assertion be re-enabled?

[salvatore-campagna]

indexRouting.createId requires a Map as input... decodeTsid does not return a Map anymore.
I don't think we can have that assertion as it was before. Note that decodeTsid returns a Base64 encoding of the tsid and I see no way to retrieve the id given a Base64 encoding of the tsid. Maybe I can do something using the plain (not base64 encoded) tsid.

[felixbarny]

Not saying it's better but one alternative idea to only hashing if the tsid is over 32k I had was to always hash but make the hash consist of two parts: ${similarityHash(_tsid)}-${hash(_tsid)}.

This would have two potential advantages:

• Always use the same hashing approach, regardless of the length of the input. This avoids having a rarely executed special case for large _tsids.
• Potentially does a better job at co-locating time series that are similar close to one another. IINM, this is desirable as it reduces disk usage. Currently, if the first dimension for two time series differ, but the rest of the dimensions are the same, they're not co-located on disk because of the lexicographically sort order based on the _tsid. Using a similarity hash as the first part of the _tsid may not have this limitation.

However, I suppose that this approach would need more testing and tuning than the approach in this PR as it has an impact on all cases, not just edge cases. Not sure what impact this would have on downsampling and its assumptions around ordering. Also, this would change the _tsid of all time series when upgrading from a previous version to that new strategy.

I'll leave it up to you and the team to decide if it's worth investigating that approach.

[salvatore-campagna]

Right now running the cardinality aggregation on one of the dimension fields fails.

[salvatore-campagna]

@elasticmachine test this please

[salvatore-campagna]

@elasticmachine run elasticsearch-ci/bwc

[salvatore-campagna]

@felixbarny do you have something specific in mind for similarityHash?

I understand the goal is to favor clustering of similar tsid but I was wondering if you are thinking of a specific hash function.

[felixbarny]

I was thinking of a locality-sensitive hash (LSH). Maybe there's already something in the ES code base that we could re-use, not sure. Maybe something related to nearest neighbor search or maybe we can reuse parts of the MinHash token filter.

I bet folks like @jpountz or @benwtrent would know best what's an appropriate hash function for that use case.

[felixbarny]

Seems like this is somewhat related: apache/lucene#12489

[salvatore-campagna]

@felixbarny I was considering doing something a bit simpler like for instance generate the first piece of the hash, hashing just a subset of the dimensions. I was considering to include in this subset at least the set of fields used in the routing_path so that colocation is also related to the documents actual physical shard location.

So something like

hash(routing_path)

putting a limit like an upper bound to the number of fields included at least to deal with the fact that, if I understand correctly, typically integrations use routing_path = dimensions.

So something more like

hash(subset(dimensions, LIMIT))

where LIMIT is some integer value lower that the total number of dimensions.

Anyway this probably would not work well if the set of fields in the routing_path is small...which means other than having an upper bound I would also need a lower bound.

[salvatore-campagna]

Another reason why I was considering what I said above is that it would allow us, maybe later, to also add an optional additional attribute to dimension fields to have applications controlling which dimensions to hash, because for instance applications know better which time series are used together?

[benwtrent]

@felixbarny @salvatore-campagna do we know the tsids ahead of time? I am guessing not.

Could we get away with only hashing PART of the tsid? What constitutes the biggest part of it?

There are various order-preserving hashes out there, I am honestly surprised we don't have one implemented in Elasticsearch yet.

I don't think you want to "cluster" things but instead you want the order of things to be preserved.

A variation on MinHashFilter could work. But I don't think it would be trivial to implement. Probably something that takes inspiration from that, or one of the order-preserving near-perfect hashes out there.

[felixbarny]

I don't think you want to "cluster" things but instead you want the order of things to be preserved.

Overall, the goal is to reduce storage space for the index data structures. As TSDB has a built-in index sorting on the _tsid field, the order of that field determines which time series are co-located to one another. The assumption is that if you co-locate _tsids that have similar dimensions, the index data structures can be more compact.

I'm not sure if we've validated that assumption. I think it would be useful to check what happens to the storage size when comparing the current implementation where the _tsid is a concatenation of all dimensions (scenario a) vs a simple hash for the full _tsid (scenario b). We'd expect to see less storage needed for the index data structure in scenario a, as in scenario b, the distribution of time series is basically random.

However, scenario a is also not perfect: If you have two time series for a metric with 10 dimensions each. If only the first dimension differs, the two time series will probably not be co-located to one another, even though the rest of the dimensions may be the same. That's where the idea with the similarity hash comes in: if we can cluster time series whose dimensions are similar to each other, we may see less storage needed for index data structures.

However, I'm not sure if the index sort is enough for that. We may also need to think about the shard _routing. But we need to be careful that this doesn't result in shard hot-spotting in case they're different sets of clusters that have an uneven distribution.

---

Having said all of that, I can propose an approach that should have similar properties to the current mechanism but allows us to remove the dimension limit. The basic idea is that we construct the _tsid similarly to today but instead of using the full keys and values for the dimensions, we're hashing each key and value individually with a short hash (for example 32 bit hashes like Adler32, Crc32c, or even String#hashCode) on the first n dimensions. N can be a relatively high number like 1024 and we'd still be well under Lucene's 32kb keyword token length limit. To avoid hash collisions, we prepend a larger hash (like 256 bit, such as sha256) at the end.

Pseudo code:

sort(dimensions)
_tsid = new ByteBuffer()
for (i = 0; i < dimensions.length && i < 1024; i++) {
  _tsid.put(hash32(dimensions[i].key))
  _tsid.put('=')
  _tsid.put(hash32(dimensions[i].value))
  _tsid.put(':')
}
_tsid.put(hash256(dimensions))

[salvatore-campagna]

I don't think you want to "cluster" things but instead you want the order of things to be preserved.

Overall, the goal is to reduce storage space for the index data structures. As TSDB has a built-in index sorting on the _tsid field, the order of that field determines which time series are co-located to one another. The assumption is that if you co-locate _tsids that have similar dimensions, the index data structures can be more compact.

I'm not sure if we've validated that assumption. I think it would be useful to check what happens to the storage size when comparing the current implementation where the _tsid is a concatenation of all dimensions (scenario a) vs a simple hash for the full _tsid (scenario b). We'd expect to see less storage needed for the index data structure in scenario a, as in scenario b, the distribution of time series is basically random.

However, scenario a is also not perfect: If you have two time series for a metric with 10 dimensions each. If only the first dimension differs, the two time series will probably not be co-located to one another, even though the rest of the dimensions may be the same. That's where the idea with the similarity hash comes in: if we can cluster time series whose dimensions are similar to each other, we may see less storage needed for index data structures.

However, I'm not sure if the index sort is enough for that. We may also need to think about the shard _routing. But we need to be careful that this doesn't result in shard hot-spotting in case they're different sets of clusters that have an uneven distribution.

Having said all of that, I can propose an approach that should have similar properties to the current mechanism but allows us to remove the dimension limit. The basic idea is that we construct the _tsid similarly to today but instead of using the full keys and values for the dimensions, we're hashing each key and value individually with a short hash (for example 32 bit hashes like Adler32, Crc32c, or even String#hashCode) on the first n dimensions. N can be a relatively high number like 1024 and we'd still be well under Lucene's 32kb keyword token length limit. To avoid hash collisions, we prepend a larger hash (like 256 bit, such as sha256) at the end.

Pseudo code:

_tsid = ""
for (i = 0; i < dimensions.length && i < 1024; i++) {
  _tsid += hash32(dimensions[i].key)
  _tsid += ':'
  _tsid += hash32(dimensions[i].value)
  _tsid += ';'
}
_tsid += hash256(dimensions)

This is similar to what I was considering above, where the subset of dimensions corresponds to the first 1024 dimensions here (LIMIT =1024).

[felixbarny]

The difference is that I was proposing to hash each dimension key and value individually (a) where I think you were proposing to take a subset of the dimensions and create a single hash (b) out of that.

In a case where only the last dimension value is different, with approach b, the hash is entirely different but with approach a, only the last part will differ. Therefore, potentially doing a better job at clustering similar time series next to one another.

But I'm still not sure whether we're chasing ghosts here and whether that clustering actually makes a big difference in storage size.

[salvatore-campagna]

The difference is that I was proposing to hash each dimension key and value individually (a) where I think you were proposing to take a subset of the dimensions and create a single hash (b) out of that.

In a case where only the last dimension value is different, with approach b, the hash is entirely different but with approach a, only the last part will differ. Therefore, potentially doing a better job at clustering similar time series next to one another.

But I'm still not sure whether we're chasing ghosts here and whether that clustering actually makes a big difference in storage size.

To answer the last part I would like to werite the PR so that to have a branch where, by changing a setting, we can test the following:

1. hash without similarity for _tsid > 32 kb
2. hash with similarity for _tsid > 32 kb
3. hash always without similarity
4. hash always with similarity