TimeSeries Proposal

[yezhizi]

This is an OSPP 2025 project. I sincerely appreciate and welcome feedback and suggestions from all community members to help me improve!

Introduction

RedisTimeSeries is a redis module used to operate and query time series data, giving redis basic time series database capabilities.

As Apache Kvrocks is characterized by being compatible with the Redis protocol and commands, we also hope to provide temporal data processing capabilities that are compatible with RedisTimeSeries.

---

Data Structure Design

Most of the design is inspired by beihao's proposal and PR #2573, but I've added many details and made significant changes to the core data storage approach.

Key Type(1 byte emun)

key type	enum value
CHUNK	0
LABEL	1
DOWNSTREAM	2

1. Datameta (Metadata CF)

        +----------+----------+-----------+------------------+----------------+-----------+-----------+----------------+----------------+-----------+----------------+
key =>  |  flags   |  expire  |  version  | size(chunkCount) | retentionTime  | chunkSize | chunkType | duplicatePolicy| sourceKey_size | sourceKey | lastTimestamp  |
        | (1byte)  |  (Ebyte) |  (8byte)  |      (Sbyte)     |    (8byte)     |  (8byte)  |  (1byte)  |    (1byte)     |    (4byte)     |  (Xbyte)  |   (8byte)      |
        +----------+----------+-----------+------------------+----------------+-----------+-----------+----------------+----------------+-----------+----------------+

• retentionTime: Maximum age (in milliseconds) for samples compared to the latest timestamp. A value of 0 disables retention.
• chunkSize: The preferred number of samples per data chunk.
• chunkType: The storage format of the chunk (compressed or uncompressed).
• duplicatePolicy: An enum represents the policy to handle samples with duplicate timestamps (e.g., BLOCK, FIRST, LAST).
• sourceKey: If this series is a downstream target for compaction, this field stores the key of the source series.
• lastTimestamp: An approximate latest timestamp in the time series.

2. Primary Data Storage (Time Chunks)

This section explains how timestamp and value data are stored. In RedisTimeSeries, both compressed and uncompressed storage methods are supported. Below is the corresponding design in KvRocks:

Uncompressed Chunk Type

Data storage (default CF):

                              +--------+-------------+-------------+-------------+-------------+
key|version|CHUNK|chunk_id => | count  |  timestamp1 |    value1   |  timestamp2 |    value2   |...
                              |(8byte) |   (8byte)   |   (8byte)   |   (8byte)   |   (8byte)   |...
                              +--------+-------------+-------------+-------------+-------------+

Uncompressed chunks store raw timestamps and values as uint64 and double types, respectively.

Compressed Chunk Type

Data storage (default CF):

                              +--------+-------------+-------------+-------------+-------------+
key|version|CHUNK|chunk_id => | count  |  timestamp1 |    value1   |  timestamp2 |    value2   |...
                              |(8byte) |   (Xbyte)   |   (Ybyte)   |   (Zbyte)   |   (Kbyte)   |...
                              +--------+-------------+-------------+-------------+-------------+

• Compression algorithm implementation (see RedisTimeSeries/src/gorilla.c):
  • timestamp: Delta-of-delta encoding.
  • value: Facebook Gorilla.

Compression might not be mandatory initially, but as noted in influxDB, it helps "reduce storage space and disk IO when querying". We may start with uncompressed chunks and add compression later.

3. Secondary Indexes

Label Index (default cf)

                                +------------------+
key|version|LABEL|label_key1 => |  label_value1    |
                                |     (Xbyte)      |  
                                +------------------+
                                +------------------+
key|version|LABEL|label_key2 => |  label_value2    |
                                |     (Xbyte)      |  
                                +------------------+
...

• Used to retrieve all labels via TS.INFO.
• Supports modifying labels with TS.ALTER

Reverse Index (TimeSeries CF)

Since querying multiple time series using label indexing is a very common operation, designing an efficient reverse index is crucial. Given that the key encoding format of the reverse index differs significantly, it is necessary to introduce a new Column Family (TimeSeries).

+-----------+--------------+----------------+------------+------------------+--------------+---------+
|     ns    |  index_type  | label_key_size |  label_key | label_value_size |  label_value |  key    | => null
| (1+Xbytes)|    (1byte)   |    (4byte)     |   (Ybyte)  |   (4byte)        |   (Zbyte)    | (Kbyte) |
+-----------+--------------+----------------+------------+------------------+--------------+---------+

• This is used to filter by labels in TS.MRANGE
• This may need to be placed in a new column family.

Downstream Key Datameta (default cf)

In Redis, you can use TS.CREATERULE to create one or more "destination timeseries" for compaction from a "source timeseries" (see here). Here we use "downstream" to represent the destination timeseries for better semantics.

                              		     +-------------+----------------+-------------+---------------------+---------------+
key|version|DOWNSTREAM|downstream_key => |  aggregator |bucket_duration |  alignment  | latest_bucket_index |  auxinfo      |
                              		     |   (1byte)   |   (8byte)      |   (8byte)   |       (8byte)       |   (XByte)     |
                                         +-------------+----------------+-------------+---------------------+---------------+

• Supports the rules field in TS.INFO.
• aggregator is an enum for aggregation methods supported by TS.CREATERULE. The meanings of aggregator, bucket_duration, and alignment are defined in the Redis command.
• latest_bucket_indexhelps quickly determine if it's the latest bucket when adding data downstream.
• auxInfo: Supports various aggregation types in Redis and helps downstream series update quickly with lower CPU overhead(see Quickly Update Latest Bucket).

Key Operation Workflows

Writing

TS.ADD key timestamp value 
  [RETENTION retentionPeriod] 
  [ENCODING <COMPRESSED|UNCOMPRESSED>] 
  [CHUNK_SIZE size] 
  [DUPLICATE_POLICY policy] 
  [ON_DUPLICATE policy_ovr] 
  [IGNORE ignoreMaxTimediff ignoreMaxValDiff] 
  [LABELS [label value ...]]

1. Get datameta and its lastTimestamp field.
2. Compute chunk_id from timestamp using: chunk_id = timestamp // chunk_size
3. Determine if it's the latest data:
   1. Yes( lastTimestamp<timestamp): Append data to the corresponding chunk(chunk_id) . For compressed types, this can quickly compute compressed values using chunk datameta and append to chunk end, reducing CPU usage; uncompressed types insert directly.
   2. No(lastTimestamp>timestamp): For uncompressed types, use binary search to locate insertion position; compressed types need to reconstruct the entire chunk.
4. Pass the data to the downstream series. See below.
5. Update datametaandchunk datameta.

Optimizations for Compressed Data Writes

There are two distinct patterns for time series writes that require different handling. For source series, new data points are typically "appended" to chunks, while downstream series primarily involve "updating" the latest data points. Compression algorithms (Delta-of-delta encoding and Facebook Gorilla) are both differential-based algorithms. We can implement different approaches tailored to these two scenarios.

1. "Append" mode: Updates prev_timestamp, prev_timestampDelta, prev_value during writes.

2. "Update" mode: prev_timestamp and prev_value point to the second-newest data, allowing direct updates of the latest data.

• The last_offset field in Downstream Key Datameta helps quickly locate the latest data point for "update" mode optimizations.
• Source series defaults to "append" mode, downstream to "update" mode.

Time Range Query (TS.RANGE)

TS.RANGE ts_key start_ts end_ts [AGGREGATION avg 60000] [FILTER_BY_VALUE 25 40]

1. Adjust start_ts to end_ts to ensure the query range hasn't expired.
2. Compute chunk_id range from start_ts to end_ts.
3. Iterate chunks using ns_key|version|*.
4. Return filtered and aggregated(if needed) results.

Multi-Series Query (TS.MRANGE)

TS.MRANGE start_ts end_ts FILTER sensor_type=temperature FILTER_BY_VALUE 0 100

1. Resolve ts_key candidates via label index. See Reverse Index
2. Execute parallel TS.RANGE on each candidate.
3. Merge results with filtering.

Passing Data Downstream (TS.CREATERULE)

Here we use the term "bucket" consistently with Redis, where data in the same bucket will be downsampled into one data point. See here. chunk refers to storage units where data in the same chunk is written to the same location. These terms have different meanings.

When a new data point is added to the source series (e.g., (ts, value)):

1. Check downstream_key and get aggregator,bucket_duration,alignment , latest_bucket_index via [Downstream Key Datameta](#Downstream Key Datameta).

2. Calculate bucket_index: (ts-align)//bucket_duration, and aligned time ts': ts' = bucket_index*bucket_duration+align

   1. bucket_index > latest_bucket_index: Directly add a new data point (ts',value) downstream.
   2. bucket_index = latest_bucket_index: Most frequent case, see Quickly Update Latest Bucket.
   3. bucket_index < latest_bucket_index：
      1. No need to fetch from source: e.g., when aggregation type is MIN and new value > current value.
      2. Need to fetch all bucket data from source and recalculate: e.g., when aggregation type is AVG.

3. Write result to downstream_key atomically, consistent with Writing.

Quickly Update Latest Bucket

This is the most common case for downstream time series data writes, corresponding to the "update" mode in Optimizations for Compressed Data Writes. The auxiliary information in Downstream Key Datameta helps determine updated values without fetching from source, reducing I/O and CPU since bucket_duration may be much larger than chunk_size.

Specifically, we store current bucket statistics for different aggregation types:

Aggregation type	Stored Information	Update Logic
AVG	Sum and count of the samples	update_avg=(cur_avg_value*count+new_value)/(count+1)
SUM	Sum of the samples	new_sum=cur_sum+new_value
MIN	Current min value	new_min_value=new_value<cur_min_value?new_value:cur_min_value
MAX	Current max value	Similar to MIN
RANGE	Current max and min value	Update cur_min_value和cur_max_value，thennew_range = cur_max_value - cur_min_value
FIRST	First timestamp	Update only if new_timestamp<first_timestamp
LAST	Latest timestamp	Similar to FIRST
std.p	Count, sum and sum of squares	See Welford's algorithm
std.s	Count, sum and sum of squares	Same as above
var.p	Count, sum and sum of squares	Same as above
var.s	Count, sum and sum of squares	Same as above

Retention ( [RETENTION retentionPeriod] )

Use RocketDB compact filters to delete expired data. When all data in a chunk expires, the chunk can be deleted (when chunk_last_timestamp+retention<lastTimestamp). First delete expired chunk datameta, then the corresponding chunk.

Queries automatically filter data outside the retention window.

Command Support Status

This section outlines the core commands.

RedisTimeSeries Commands	Optional arguments or output fields	Supported?
TS.ADD/TS.MADD	[DUPLICATE_POLICY policy]	✅
TS.CREATE	[ON_DUPLICATE policy_ovr]	✅
	[LABELS [label value ...]]	✅
	[RETENTION retentionPeriod]	✅
	[ENCODING]	✅
	[CHUNK_SIZE size]	❓
	[LABELS [label value ...]]	✅
TS.DEL		✅
TS.MRANGE/TS.RANGE	[FILTER_BY_TS ts...]	✅
	[FILTER_BY_VALUE min max]	✅
	[COUNT count]	✅
	[[ALIGN align] AGGREGATION aggregator bucketDuration [BUCKETTIMESTAMP bt] [EMPTY]]	✅
	FILTER filterExpr...	✅
TS.CREATERULE	AGGREGATION aggregator bucketDuration	✅
	[alignTimestamp]	✅
TS.INCRBY/TS.DECRBY		✅
TS.INFO	memoryUsage	❓
	firstTimestamp	✅
	chunkSize	❓
	other...	✅

❓indicates fields that may be incompatible or inconsistent with Redis implementation.

TS.INFO Field Compatibility

Some fields may not align with our design:

• chunkSize: In Redis, this is a fixed byte size specified at creation.
• memoryUsage: This metric might be challenging to calculate and maintain accurately. As an alternative, we could potentially use diskUsage instead, with support from the chunk datameta structure ?

[yezhizi]

cc @PragmaTwice

[PragmaTwice]

Thank you for your great proposal!

Feel free to comment, ask questions and give suggestions about this proposal. cc @apache/kvrocks-committers

[PragmaTwice]

Some comments:

• In Kvrocks there are several existing column families. And the most relevant CFs here are metadata (metadata CF) and default (subkey CF), where all keys inside should follow a special key format (user keys and sub keys, per https://kvrocks.apache.org/community/data-structure-on-rocksdb#user-key-encoding). I think it is clear that "Datameta" in the proposal is stored in metadata CF, but for other parts we can specify clearly whether it is stored in subkey CF or a new timeseries CF.
• There may be some pros and cons for introducing a new CF (e.g. we face some compatibility issue of introducing one). If we introduce a CF for timeseries, we should consider:
  • How to avoid any key conflicts? e.g. in the proposal, there may be some conflicts since we didn't introduce some enum values as prefixes in the key format (for example, a key in "Reverse Index" format may be same as another key in "Label Index"). Also, we cannot iterate over a special type of keys if we don't distingush them by prefixing keys with some enum values (recall that rocksdb key-values are sorted in lexicographical order). (Maybe you can find some idea in the encoding of search data: https://kvrocks.apache.org/community/kvrocks-search-index-encoding.)
  • How to do garbage-collection? We have key expiration and also timeseries-specific "retention", and we need to modify compaction filter (https://github.com/apache/kvrocks/blob/unstable/src/storage/compact_filter.cc) and maybe also compaction checker (https://github.com/apache/kvrocks/blob/unstable/src/storage/compaction_checker.cc) to figure out (1): which key is expired if a compaction is triggered, and (2): when we need to trigger a compaction (maybe there's timeseries specific logic).
• Also I found some reference to codebase of redis timeseries. We can of course read its code, but we cannot use them directly (i.e. copy ANY code from its codebase) due to ASF license requirements (it's AGPLv3 licensed).

[PragmaTwice]

Yeah for example, if a request TYPE k is received by Kvrocks, it cannot search all CFs to get the type for k, instead it will check if it exists in the metadata CF, i.e. there is a one-to-one mapping from redis keys to rocksdb keys (metadata) in the metadata CF.

For some types like HASH, there is a one-to-many mapping from redis keys to rocksdb keys (e.g. fields in a HASH) in the default (subkey) CF.

[git-hulk]

Thanks for your great proposal.

For the encoding part, I think it would be better to encode the key type instead of introducing a new column family, which will allow users to rollback if have any issues in newer versions.

               +------------------+----------------+------------------+------------------+
key|version => |  label_key1_size |  label_key1    |label_value1_size |  label_value1    |
               |     (4byte)      |   (Xbyte)      |   (4byte)        |     (Ybyte)      | 
               +------------------+----------------+------------------+------------------+
               +------------------+----------------+------------------+------------------+
               |  label_key2_size |  label_key2    |label_value2_size |  label_value2    |...
               |     (4byte)      |   (Zbyte)      |   (4byte)        |     (Sbyte)      |...
               +------------------+----------------+------------------+------------------+

For the secondary index, do you mean that it will put all of label indexes in one key-value?

[yezhizi]

Thanks a lot for your suggestions!

1. OK. I will revised the key format design and try key type encoding.
2. Yes, that's currently the case. All labels for a specific key are stored in a single key-value. Are there any issues or suggestions?

[git-hulk]

Yes, that's currently the case. All labels for a specific key are stored in a single key-value. Are there any issues or suggestions?

I see. My concern is it might has the performance issue since one time series might have thousand label values. And we need to read-modify-write this value every time a new label was added.

[yezhizi]

I see. My concern is it might has the performance issue since one time series might have thousand label values. And we need to read-modify-write this value every time a new label was added.

OK, splitting it into multiple keys might be more reasonable. When users update labels, we can modify only the relevant keys.
Something like that:

                          +------------------+
key|version|label_key1 => |  label_value1    |
                          |     (Xbyte)      |  
                          +------------------+
                          +------------------+
key|version|label_key2 => |  label_value2    |
                          |     (Xbyte)      |  
                          +------------------+
...

[xiaobiaozhao]

I have a few questions

1. I don't see how you designed to persist this data into rocksdb.
2. I think you can consider the following: Is it possible to take advantage of the natural ordering of the rocksdb data store to store this temporal structure? eg. rocksdb' seek func.

[yezhizi]

Thank you for your suggestions and feedback.

1. Actually, in our current proposal design, we organize timestamp and value data into chunks("2. Primary Data Storage (Time Chunks)"), which are then persisted as key-value pairs in RocksDB.
2. Thanks for your suggestion! Our proposal will use chunk_id as part of the RocksDB key for indexing. Each chunk_id represents a specific time window (e.g., 1 minute for chunk_size=60000ms), with all timestamps falling within that window stored in the corresponding chunk. This enables us to use the Seek function to locate time-series data within specific ranges(since most time-series operations involve range time queries).

[mapleFU]

Sorry for late replying because I'm a bit busy these days. What would agg does when TS.DEL happens?

[yezhizi]

Our proposal doesn't include specific optimizations for deletion or modification operations (TS.INCRBY, TS.DEL). When users delete or modify source series, we needs to read all data in the corresponding bucket from source series, recalculate the aggregated values, and update them in downstream series.

This may incur significant overhead. However, given that time-series data is typically 'append-only' in nature, operations that modify or delete already-written data are relatively uncommon, so I believe this is reasonable. Below is InfluxDB's description regarding time-series characteristics:

Immutability – Since time series data comes in time order, it is almost always recorded in a new entry, and as such, should be immutable and append-only (appended to the existing data). It doesn’t usually change but is rather tacked on in the order that events happen.

[mapleFU]

(Sorry again for late replying, I'll take a careful pass when I have time, now I only have time to read this when I'm not working...) I have a quick first scan at the spec, and there're some thoughts:

1. In Secondary Indexes part, the key is key|version, would this by design to hack like this?
2. Assume time series is a write heavy structure, is firstTimestamp and endTimestamp in root meta a good idea? It will update the root and updated chunk evertime a write comes. Can we Assume ts.info doesn't called frequently, and querying it in chunks ( this is not a good idea when we have too many chunks, this might also requires (3) ), or more hacking, leave first end time stamp for chunks except the last chunk ( which might be frequently appending), and querying the last chunk's end.
3. I don't know fixed time-window chunk ( I mean, chunk_id = timestamp // chunk_size , chunk_size=60000ms for minute-level chunks) a good idea. As time flies, chunks would be extremly huge. And it's really hard to tunning. Personnaly I prefer some internal merging rules to "compact" them, which requires a "dynamic" chunking
4. What does memoryUsage means in metadata? I think actually it's hard to estimate "memoryUsage", since kvrocks currently doesn't have a explicit "cache" layer.
5. I don't clear how Compressed Chunk Type works, the ts1 value1 ts2 value2 is a bit hacking, why not using ts1 ts2 ... for times, and value1, value2 for values? Spliting to one stream might not so good for size-compression ( because of data locality ), but might friendly for appending data ( in fact ts1 value1 ts2 value2 with byte-level compression doesn't need to rebuild data structure when appending?)

Thanks again for the proposal

[mapleFU]

inserting algorithm (insert ts1, v1):

Finally I have time for this. I think the key design looks good to me. The problem is focus on the insert. Basically my idea is:

Terminology:

1. Sealed Chunk: A chunk, which could be compressed, and likely to be queryed and not inserted frequently. It can still be updated, but we don't regard update into it a good behavior
2. Unsealed Chunk: A chunk that is likely to be inserted, it should be "uncompressed"

The chunks layouts like:

[ sealed ] [ sealed ] [ sealed ] [unsealed ...)

When insert:

1. if it's inserting to unsealed chunk, just append into it. If unsealed chunk size reaches chunk-size, switch a new chunk, and seals this chunk ( or maybe seal it later )
2. If it's a sealed chunk, we need overwrite the whole chunk, and rewrite it. We can supports split, but this is not neccessary here.

Note that this don't follows 50% ratio. Because it's likely to append to the "newest" unsealed chunk.

[mapleFU]

Another problem is kvrocks is not design for append heavy behavior. For detail, the unsealed chunk would be overwrite and overwrite again, causing high amplify ratio. We can:

1. Let user make their best to call MADD
2. (maybe better way, but need refactor) Given the ability to "batch" inserting. When there're high ADD qps, we can grouping them to batching the handling
3. (maybe a bit hard) allow lightweight update for unsealed chunk, like write a small key-value or small value for "unsealed" chunk, and merging them when sealing.

Personally I may prefer (1) and (2)

[mapleFU]

(And personally, maybe "Merge" Operator in rocksdb is also worth to try in this scenerio, see https://github.com/facebook/rocksdb/wiki/Merge-Operator , we can regard it as an optimization when we benchmarking)

[yezhizi]

Sorry for the late reply. Thank you all so much for your suggestions and the detailed design—this is really helpful!

1. The design looks great to me, and I’ll start by implementing a "basic version" based on it.

chunk size M (order of the "B-tree")

chunk_id: first timestamp in the chunk

user_key | CHUNK | chunk_id => count | ts1 | v1 | ... | tsN | vN

2. The idea of treating chunks as sealed and unsealed is great. It can also simplify many operations, including writes to downstream.

[ sealed ] [ sealed ] [ sealed ] [unsealed ...)

3. I’ll begin working on some code submissions soon. Thanks again for your input!

[mapleFU]

Yeah, I think this way is easier and working on some write heavy scenerio. A final problem I talked previously is that, the "newest" chunk might be rewrite and rewrite(and might read) again during user operations. Maybe we can think about how to handle this after basic structure is implemented.