Skip to content

Commit cc82eea

Browse files
yezhiziPragmaTwice
yezhizi
and
PragmaTwice
authored
feat(ts): initialize metadata and subkey encoding (#3072)
Signed-off-by: DeEMO <yzzxrx@gmail.com> Co-authored-by: Twice <twice@apache.org>
1 parent 6719235 commit cc82eea

4 files changed

Lines changed: 301 additions & 3 deletions

File tree

‎src/storage/redis_metadata.cc‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -537,3 +537,33 @@ rocksdb::Status TDigestMetadata::Decode(Slice *input) {
537537

538538
return rocksdb::Status::OK();
539539
}
540+
541+
void TimeSeriesMetadata::SetSourceKey(Slice key) { source_key = key.ToString(); }
542+
543+
void TimeSeriesMetadata::Encode(std::string *dst) const {
544+
Metadata::Encode(dst);
545+
PutFixed64(dst, retention_time);
546+
PutFixed64(dst, chunk_size);
547+
PutFixed8(dst, static_cast<uint8_t>(chunk_type));
548+
PutFixed8(dst, static_cast<uint8_t>(duplicate_policy));
549+
PutSizedString(dst, source_key);
550+
}
551+
552+
rocksdb::Status TimeSeriesMetadata::Decode(Slice *input) {
553+
if (auto s = Metadata::Decode(input); !s.ok()) {
554+
return s;
555+
}
556+
if (input->size() < sizeof(uint64_t) * 2 + sizeof(uint8_t) * 2 + sizeof(uint32_t)) {
557+
return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
558+
}
559+
560+
GetFixed64(input, &retention_time);
561+
GetFixed64(input, &chunk_size);
562+
GetFixed8(input, reinterpret_cast<uint8_t *>(&chunk_type));
563+
GetFixed8(input, reinterpret_cast<uint8_t *>(&duplicate_policy));
564+
Slice source_key_slice;
565+
GetSizedString(input, &source_key_slice);
566+
source_key = source_key_slice.ToString();
567+
568+
return rocksdb::Status::OK();
569+
}

‎src/storage/redis_metadata.h‎

Lines changed: 46 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,7 @@ enum RedisType : uint8_t {
5353
kRedisJson = 10,
5454
kRedisHyperLogLog = 11,
5555
kRedisTDigest = 12,
56+
kRedisTimeSeries = 13,
5657
};
5758

5859
struct RedisTypes {
@@ -94,9 +95,9 @@ enum RedisCommand {
9495
kRedisCmdLMove,
9596
};
9697

97-
const std::vector<std::string> RedisTypeNames = {"none", "string", "hash", "list", "set",
98-
"zset", "bitmap", "sortedint", "stream", "MBbloom--",
99-
"ReJSON-RL", "hyperloglog", "TDIS-TYPE"};
98+
const std::vector<std::string> RedisTypeNames = {"none", "string", "hash", "list", "set",
99+
"zset", "bitmap", "sortedint", "stream", "MBbloom--",
100+
"ReJSON-RL", "hyperloglog", "TDIS-TYPE", "timeseries"};
100101

101102
constexpr const char *kErrMsgWrongType = "WRONGTYPE Operation against a key holding the wrong kind of value";
102103
constexpr const char *kErrMsgKeyExpired = "the key was expired";
@@ -364,3 +365,45 @@ class TDigestMetadata : public Metadata {
364365

365366
double Delta() const { return 1. / static_cast<double>(compression); }
366367
};
368+
369+
class TimeSeriesMetadata : public Metadata {
370+
public:
371+
enum class ChunkType : uint8_t {
372+
UNCOMPRESSED = 0,
373+
COMPRESSED = 1,
374+
};
375+
376+
enum class DuplicatePolicy : uint8_t {
377+
BLOCK = 0,
378+
FIRST = 1,
379+
LAST = 2,
380+
MIN = 3,
381+
MAX = 4,
382+
SUM = 5,
383+
};
384+
385+
uint64_t retention_time;
386+
uint64_t chunk_size;
387+
ChunkType chunk_type;
388+
DuplicatePolicy duplicate_policy;
389+
std::string source_key;
390+
391+
explicit TimeSeriesMetadata(bool generate_version = true)
392+
: Metadata(kRedisTimeSeries, generate_version),
393+
retention_time(0),
394+
chunk_size(0),
395+
chunk_type(ChunkType::UNCOMPRESSED),
396+
duplicate_policy(DuplicatePolicy::BLOCK) {}
397+
TimeSeriesMetadata(uint64_t retention_time, uint64_t chunk_size, ChunkType chunk_type,
398+
DuplicatePolicy duplicate_policy, bool generate_version = true)
399+
: Metadata(kRedisTimeSeries, generate_version),
400+
retention_time(retention_time),
401+
chunk_size(chunk_size),
402+
chunk_type(chunk_type),
403+
duplicate_policy(duplicate_policy) {}
404+
405+
void SetSourceKey(Slice key);
406+
407+
void Encode(std::string *dst) const override;
408+
rocksdb::Status Decode(Slice *input) override;
409+
};

‎src/types/redis_timeseries.cc‎

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,122 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*
19+
*/
20+
21+
#include "redis_timeseries.h"
22+
23+
namespace redis {
24+
25+
void TSDownStreamMeta::Encode(std::string *dst) const {
26+
PutFixed8(dst, static_cast<uint8_t>(aggregator));
27+
PutFixed64(dst, bucket_duration);
28+
PutFixed64(dst, alignment);
29+
PutFixed64(dst, latest_bucket_idx);
30+
PutFixed8(dst, static_cast<uint8_t>(u64_auxs.size()));
31+
PutFixed8(dst, static_cast<uint8_t>(f64_auxs.size()));
32+
for (const auto &aux : u64_auxs) {
33+
PutFixed64(dst, aux);
34+
}
35+
for (const auto &aux : f64_auxs) {
36+
PutDouble(dst, aux);
37+
}
38+
}
39+
40+
rocksdb::Status TSDownStreamMeta::Decode(Slice *input) {
41+
if (input->size() < sizeof(uint8_t) * 3 + sizeof(uint64_t) * 3) {
42+
return rocksdb::Status::InvalidArgument("TSDownStreamMeta size is too short");
43+
}
44+
45+
GetFixed8(input, reinterpret_cast<uint8_t *>(&aggregator));
46+
GetFixed64(input, &bucket_duration);
47+
GetFixed64(input, &alignment);
48+
GetFixed64(input, &latest_bucket_idx);
49+
uint8_t u64_auxs_size = 0;
50+
GetFixed8(input, &u64_auxs_size);
51+
uint8_t f64_auxs_size = 0;
52+
GetFixed8(input, &f64_auxs_size);
53+
54+
// Strict checking to prevent accidental overwrites
55+
if (input->size() != sizeof(uint64_t) * u64_auxs_size + sizeof(double) * f64_auxs_size) {
56+
return rocksdb::Status::InvalidArgument("Invalid auxinfo size");
57+
}
58+
59+
for (uint8_t i = 0; i < u64_auxs_size; i++) {
60+
uint64_t aux = 0;
61+
GetFixed64(input, &aux);
62+
u64_auxs.push_back(aux);
63+
}
64+
for (uint8_t i = 0; i < f64_auxs_size; i++) {
65+
double aux = 0;
66+
GetDouble(input, &aux);
67+
f64_auxs.push_back(aux);
68+
}
69+
70+
return rocksdb::Status::OK();
71+
}
72+
73+
std::string TSRevLabelKey::Encode() const {
74+
std::string encoded;
75+
size_t total = 1 + ns.size() + 1 + 4 + label_key.size() + 4 + label_value.size() + user_key.size();
76+
77+
encoded.resize(total);
78+
auto buf = encoded.data();
79+
buf = EncodeFixed8(buf, static_cast<uint8_t>(ns.size()));
80+
buf = EncodeBuffer(buf, ns);
81+
buf = EncodeFixed8(buf, static_cast<uint8_t>(IndexKeyType::TS_LABEL));
82+
buf = EncodeFixed32(buf, static_cast<uint32_t>(label_key.size()));
83+
buf = EncodeBuffer(buf, label_key);
84+
buf = EncodeFixed32(buf, static_cast<uint32_t>(label_value.size()));
85+
buf = EncodeBuffer(buf, label_value);
86+
EncodeBuffer(buf, user_key);
87+
88+
return encoded;
89+
}
90+
91+
std::string TimeSeries::internalKeyFromChunkID(const std::string &ns_key, const TimeSeriesMetadata &metadata,
92+
uint64_t id) const {
93+
std::string sub_key;
94+
PutFixed8(&sub_key, static_cast<uint8_t>(TSSubkeyType::CHUNK));
95+
PutFixed64(&sub_key, id);
96+
97+
return InternalKey(ns_key, sub_key, metadata.version, storage_->IsSlotIdEncoded()).Encode();
98+
}
99+
100+
std::string TimeSeries::internalKeyFromLabelKey(const std::string &ns_key, const TimeSeriesMetadata &metadata,
101+
Slice label_key) const {
102+
std::string sub_key;
103+
sub_key.resize(1 + label_key.size());
104+
auto buf = sub_key.data();
105+
buf = EncodeFixed8(buf, static_cast<uint8_t>(TSSubkeyType::LABEL));
106+
EncodeBuffer(buf, label_key);
107+
108+
return InternalKey(ns_key, sub_key, metadata.version, storage_->IsSlotIdEncoded()).Encode();
109+
}
110+
111+
std::string TimeSeries::internalKeyFromDownstreamKey(const std::string &ns_key, const TimeSeriesMetadata &metadata,
112+
Slice downstream_key) const {
113+
std::string sub_key;
114+
sub_key.resize(1 + downstream_key.size());
115+
auto buf = sub_key.data();
116+
buf = EncodeFixed8(buf, static_cast<uint8_t>(TSSubkeyType::DOWNSTREAM));
117+
EncodeBuffer(buf, downstream_key);
118+
119+
return InternalKey(ns_key, sub_key, metadata.version, storage_->IsSlotIdEncoded()).Encode();
120+
}
121+
122+
} // namespace redis

‎src/types/redis_timeseries.h‎

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,103 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*
19+
*/
20+
21+
#pragma once
22+
23+
#include <cstdint>
24+
25+
#include "storage/redis_db.h"
26+
#include "storage/redis_metadata.h"
27+
28+
namespace redis {
29+
30+
enum class TSSubkeyType : uint8_t {
31+
CHUNK = 0,
32+
LABEL = 1,
33+
DOWNSTREAM = 2,
34+
};
35+
36+
// Enum prefix for new CF.
37+
enum class IndexKeyType : uint8_t {
38+
TS_LABEL = 0,
39+
};
40+
41+
enum class TSAggregatorType : uint8_t {
42+
AVG = 0,
43+
SUM = 1,
44+
MIN = 2,
45+
MAX = 3,
46+
RANGE = 4,
47+
COUNT = 5,
48+
FIRST = 6,
49+
LAST = 7,
50+
STD_P = 8,
51+
STD_S = 9,
52+
VAR_P = 10,
53+
VAR_S = 11,
54+
};
55+
56+
struct TSDownStreamMeta {
57+
TSAggregatorType aggregator;
58+
uint64_t bucket_duration;
59+
uint64_t alignment;
60+
uint64_t latest_bucket_idx;
61+
62+
// store auxiliary info for each aggregator.
63+
// e.g. for avg, need to store sum and count: u64_auxs={count}, f64_auxs={sum}
64+
std::vector<uint64_t> u64_auxs;
65+
std::vector<double> f64_auxs;
66+
67+
TSDownStreamMeta() = default;
68+
TSDownStreamMeta(TSAggregatorType aggregator, uint64_t bucket_duration, uint64_t alignment,
69+
uint64_t latest_bucket_idx)
70+
: aggregator(aggregator),
71+
bucket_duration(bucket_duration),
72+
alignment(alignment),
73+
latest_bucket_idx(latest_bucket_idx) {}
74+
75+
void Encode(std::string *dst) const;
76+
rocksdb::Status Decode(Slice *input);
77+
};
78+
79+
struct TSRevLabelKey {
80+
Slice ns;
81+
Slice label_key;
82+
Slice label_value;
83+
Slice user_key;
84+
85+
TSRevLabelKey(Slice ns, Slice label_key, Slice label_value, Slice user_key = Slice())
86+
: ns(ns), label_key(label_key), label_value(label_value), user_key(user_key) {}
87+
88+
[[nodiscard]] std::string Encode() const;
89+
};
90+
91+
class TimeSeries : public SubKeyScanner {
92+
public:
93+
TimeSeries(engine::Storage *storage, const std::string &ns) : SubKeyScanner(storage, ns) {}
94+
95+
private:
96+
std::string internalKeyFromChunkID(const std::string &ns_key, const TimeSeriesMetadata &metadata, uint64_t id) const;
97+
std::string internalKeyFromLabelKey(const std::string &ns_key, const TimeSeriesMetadata &metadata,
98+
Slice label_key) const;
99+
std::string internalKeyFromDownstreamKey(const std::string &ns_key, const TimeSeriesMetadata &metadata,
100+
Slice downstream_key) const;
101+
};
102+
103+
} // namespace redis

0 commit comments

Comments
 (0)