1 // Copyright (c) 2017-present, Facebook, Inc. All rights reserved.
2 // This source code is licensed under both the GPLv2 (found in the
3 // COPYING file in the root directory) and Apache 2.0 License
4 // (found in the LICENSE.Apache file in the root directory).
5
6 #include "format.h"
7
8 #include <algorithm>
9 #include <map>
10 #include <memory>
11
12 #include "utilities/cassandra/serialize.h"
13
14 namespace ROCKSDB_NAMESPACE {
15 namespace cassandra {
16 namespace {
17 const int32_t kDefaultLocalDeletionTime =
18 std::numeric_limits<int32_t>::max();
19 const int64_t kDefaultMarkedForDeleteAt =
20 std::numeric_limits<int64_t>::min();
21 }
22
ColumnBase(int8_t mask,int8_t index)23 ColumnBase::ColumnBase(int8_t mask, int8_t index)
24 : mask_(mask), index_(index) {}
25
Size() const26 std::size_t ColumnBase::Size() const {
27 return sizeof(mask_) + sizeof(index_);
28 }
29
Mask() const30 int8_t ColumnBase::Mask() const {
31 return mask_;
32 }
33
Index() const34 int8_t ColumnBase::Index() const {
35 return index_;
36 }
37
Serialize(std::string * dest) const38 void ColumnBase::Serialize(std::string* dest) const {
39 ROCKSDB_NAMESPACE::cassandra::Serialize<int8_t>(mask_, dest);
40 ROCKSDB_NAMESPACE::cassandra::Serialize<int8_t>(index_, dest);
41 }
42
Deserialize(const char * src,std::size_t offset)43 std::shared_ptr<ColumnBase> ColumnBase::Deserialize(const char* src,
44 std::size_t offset) {
45 int8_t mask = ROCKSDB_NAMESPACE::cassandra::Deserialize<int8_t>(src, offset);
46 if ((mask & ColumnTypeMask::DELETION_MASK) != 0) {
47 return Tombstone::Deserialize(src, offset);
48 } else if ((mask & ColumnTypeMask::EXPIRATION_MASK) != 0) {
49 return ExpiringColumn::Deserialize(src, offset);
50 } else {
51 return Column::Deserialize(src, offset);
52 }
53 }
54
Column(int8_t mask,int8_t index,int64_t timestamp,int32_t value_size,const char * value)55 Column::Column(
56 int8_t mask,
57 int8_t index,
58 int64_t timestamp,
59 int32_t value_size,
60 const char* value
61 ) : ColumnBase(mask, index), timestamp_(timestamp),
62 value_size_(value_size), value_(value) {}
63
Timestamp() const64 int64_t Column::Timestamp() const {
65 return timestamp_;
66 }
67
Size() const68 std::size_t Column::Size() const {
69 return ColumnBase::Size() + sizeof(timestamp_) + sizeof(value_size_)
70 + value_size_;
71 }
72
Serialize(std::string * dest) const73 void Column::Serialize(std::string* dest) const {
74 ColumnBase::Serialize(dest);
75 ROCKSDB_NAMESPACE::cassandra::Serialize<int64_t>(timestamp_, dest);
76 ROCKSDB_NAMESPACE::cassandra::Serialize<int32_t>(value_size_, dest);
77 dest->append(value_, value_size_);
78 }
79
Deserialize(const char * src,std::size_t offset)80 std::shared_ptr<Column> Column::Deserialize(const char *src,
81 std::size_t offset) {
82 int8_t mask = ROCKSDB_NAMESPACE::cassandra::Deserialize<int8_t>(src, offset);
83 offset += sizeof(mask);
84 int8_t index = ROCKSDB_NAMESPACE::cassandra::Deserialize<int8_t>(src, offset);
85 offset += sizeof(index);
86 int64_t timestamp =
87 ROCKSDB_NAMESPACE::cassandra::Deserialize<int64_t>(src, offset);
88 offset += sizeof(timestamp);
89 int32_t value_size =
90 ROCKSDB_NAMESPACE::cassandra::Deserialize<int32_t>(src, offset);
91 offset += sizeof(value_size);
92 return std::make_shared<Column>(
93 mask, index, timestamp, value_size, src + offset);
94 }
95
ExpiringColumn(int8_t mask,int8_t index,int64_t timestamp,int32_t value_size,const char * value,int32_t ttl)96 ExpiringColumn::ExpiringColumn(
97 int8_t mask,
98 int8_t index,
99 int64_t timestamp,
100 int32_t value_size,
101 const char* value,
102 int32_t ttl
103 ) : Column(mask, index, timestamp, value_size, value),
104 ttl_(ttl) {}
105
Size() const106 std::size_t ExpiringColumn::Size() const {
107 return Column::Size() + sizeof(ttl_);
108 }
109
Serialize(std::string * dest) const110 void ExpiringColumn::Serialize(std::string* dest) const {
111 Column::Serialize(dest);
112 ROCKSDB_NAMESPACE::cassandra::Serialize<int32_t>(ttl_, dest);
113 }
114
TimePoint() const115 std::chrono::time_point<std::chrono::system_clock> ExpiringColumn::TimePoint() const {
116 return std::chrono::time_point<std::chrono::system_clock>(std::chrono::microseconds(Timestamp()));
117 }
118
Ttl() const119 std::chrono::seconds ExpiringColumn::Ttl() const {
120 return std::chrono::seconds(ttl_);
121 }
122
Expired() const123 bool ExpiringColumn::Expired() const {
124 return TimePoint() + Ttl() < std::chrono::system_clock::now();
125 }
126
ToTombstone() const127 std::shared_ptr<Tombstone> ExpiringColumn::ToTombstone() const {
128 auto expired_at = (TimePoint() + Ttl()).time_since_epoch();
129 int32_t local_deletion_time = static_cast<int32_t>(
130 std::chrono::duration_cast<std::chrono::seconds>(expired_at).count());
131 int64_t marked_for_delete_at =
132 std::chrono::duration_cast<std::chrono::microseconds>(expired_at).count();
133 return std::make_shared<Tombstone>(
134 static_cast<int8_t>(ColumnTypeMask::DELETION_MASK),
135 Index(),
136 local_deletion_time,
137 marked_for_delete_at);
138 }
139
Deserialize(const char * src,std::size_t offset)140 std::shared_ptr<ExpiringColumn> ExpiringColumn::Deserialize(
141 const char *src,
142 std::size_t offset) {
143 int8_t mask = ROCKSDB_NAMESPACE::cassandra::Deserialize<int8_t>(src, offset);
144 offset += sizeof(mask);
145 int8_t index = ROCKSDB_NAMESPACE::cassandra::Deserialize<int8_t>(src, offset);
146 offset += sizeof(index);
147 int64_t timestamp =
148 ROCKSDB_NAMESPACE::cassandra::Deserialize<int64_t>(src, offset);
149 offset += sizeof(timestamp);
150 int32_t value_size =
151 ROCKSDB_NAMESPACE::cassandra::Deserialize<int32_t>(src, offset);
152 offset += sizeof(value_size);
153 const char* value = src + offset;
154 offset += value_size;
155 int32_t ttl = ROCKSDB_NAMESPACE::cassandra::Deserialize<int32_t>(src, offset);
156 return std::make_shared<ExpiringColumn>(
157 mask, index, timestamp, value_size, value, ttl);
158 }
159
Tombstone(int8_t mask,int8_t index,int32_t local_deletion_time,int64_t marked_for_delete_at)160 Tombstone::Tombstone(
161 int8_t mask,
162 int8_t index,
163 int32_t local_deletion_time,
164 int64_t marked_for_delete_at
165 ) : ColumnBase(mask, index), local_deletion_time_(local_deletion_time),
166 marked_for_delete_at_(marked_for_delete_at) {}
167
Timestamp() const168 int64_t Tombstone::Timestamp() const {
169 return marked_for_delete_at_;
170 }
171
Size() const172 std::size_t Tombstone::Size() const {
173 return ColumnBase::Size() + sizeof(local_deletion_time_)
174 + sizeof(marked_for_delete_at_);
175 }
176
Serialize(std::string * dest) const177 void Tombstone::Serialize(std::string* dest) const {
178 ColumnBase::Serialize(dest);
179 ROCKSDB_NAMESPACE::cassandra::Serialize<int32_t>(local_deletion_time_, dest);
180 ROCKSDB_NAMESPACE::cassandra::Serialize<int64_t>(marked_for_delete_at_, dest);
181 }
182
Collectable(int32_t gc_grace_period_in_seconds) const183 bool Tombstone::Collectable(int32_t gc_grace_period_in_seconds) const {
184 auto local_deleted_at = std::chrono::time_point<std::chrono::system_clock>(
185 std::chrono::seconds(local_deletion_time_));
186 auto gc_grace_period = std::chrono::seconds(gc_grace_period_in_seconds);
187 return local_deleted_at + gc_grace_period < std::chrono::system_clock::now();
188 }
189
Deserialize(const char * src,std::size_t offset)190 std::shared_ptr<Tombstone> Tombstone::Deserialize(const char *src,
191 std::size_t offset) {
192 int8_t mask = ROCKSDB_NAMESPACE::cassandra::Deserialize<int8_t>(src, offset);
193 offset += sizeof(mask);
194 int8_t index = ROCKSDB_NAMESPACE::cassandra::Deserialize<int8_t>(src, offset);
195 offset += sizeof(index);
196 int32_t local_deletion_time =
197 ROCKSDB_NAMESPACE::cassandra::Deserialize<int32_t>(src, offset);
198 offset += sizeof(int32_t);
199 int64_t marked_for_delete_at =
200 ROCKSDB_NAMESPACE::cassandra::Deserialize<int64_t>(src, offset);
201 return std::make_shared<Tombstone>(
202 mask, index, local_deletion_time, marked_for_delete_at);
203 }
204
RowValue(int32_t local_deletion_time,int64_t marked_for_delete_at)205 RowValue::RowValue(int32_t local_deletion_time, int64_t marked_for_delete_at)
206 : local_deletion_time_(local_deletion_time),
207 marked_for_delete_at_(marked_for_delete_at), columns_(),
208 last_modified_time_(0) {}
209
RowValue(Columns columns,int64_t last_modified_time)210 RowValue::RowValue(Columns columns,
211 int64_t last_modified_time)
212 : local_deletion_time_(kDefaultLocalDeletionTime),
213 marked_for_delete_at_(kDefaultMarkedForDeleteAt),
214 columns_(std::move(columns)), last_modified_time_(last_modified_time) {}
215
Size() const216 std::size_t RowValue::Size() const {
217 std::size_t size = sizeof(local_deletion_time_)
218 + sizeof(marked_for_delete_at_);
219 for (const auto& column : columns_) {
220 size += column -> Size();
221 }
222 return size;
223 }
224
LastModifiedTime() const225 int64_t RowValue::LastModifiedTime() const {
226 if (IsTombstone()) {
227 return marked_for_delete_at_;
228 } else {
229 return last_modified_time_;
230 }
231 }
232
IsTombstone() const233 bool RowValue::IsTombstone() const {
234 return marked_for_delete_at_ > kDefaultMarkedForDeleteAt;
235 }
236
Serialize(std::string * dest) const237 void RowValue::Serialize(std::string* dest) const {
238 ROCKSDB_NAMESPACE::cassandra::Serialize<int32_t>(local_deletion_time_, dest);
239 ROCKSDB_NAMESPACE::cassandra::Serialize<int64_t>(marked_for_delete_at_, dest);
240 for (const auto& column : columns_) {
241 column -> Serialize(dest);
242 }
243 }
244
RemoveExpiredColumns(bool * changed) const245 RowValue RowValue::RemoveExpiredColumns(bool* changed) const {
246 *changed = false;
247 Columns new_columns;
248 for (auto& column : columns_) {
249 if(column->Mask() == ColumnTypeMask::EXPIRATION_MASK) {
250 std::shared_ptr<ExpiringColumn> expiring_column =
251 std::static_pointer_cast<ExpiringColumn>(column);
252
253 if(expiring_column->Expired()){
254 *changed = true;
255 continue;
256 }
257 }
258
259 new_columns.push_back(column);
260 }
261 return RowValue(std::move(new_columns), last_modified_time_);
262 }
263
ConvertExpiredColumnsToTombstones(bool * changed) const264 RowValue RowValue::ConvertExpiredColumnsToTombstones(bool* changed) const {
265 *changed = false;
266 Columns new_columns;
267 for (auto& column : columns_) {
268 if(column->Mask() == ColumnTypeMask::EXPIRATION_MASK) {
269 std::shared_ptr<ExpiringColumn> expiring_column =
270 std::static_pointer_cast<ExpiringColumn>(column);
271
272 if(expiring_column->Expired()) {
273 std::shared_ptr<Tombstone> tombstone = expiring_column->ToTombstone();
274 new_columns.push_back(tombstone);
275 *changed = true;
276 continue;
277 }
278 }
279 new_columns.push_back(column);
280 }
281 return RowValue(std::move(new_columns), last_modified_time_);
282 }
283
RemoveTombstones(int32_t gc_grace_period) const284 RowValue RowValue::RemoveTombstones(int32_t gc_grace_period) const {
285 Columns new_columns;
286 for (auto& column : columns_) {
287 if (column->Mask() == ColumnTypeMask::DELETION_MASK) {
288 std::shared_ptr<Tombstone> tombstone =
289 std::static_pointer_cast<Tombstone>(column);
290
291 if (tombstone->Collectable(gc_grace_period)) {
292 continue;
293 }
294 }
295
296 new_columns.push_back(column);
297 }
298 return RowValue(std::move(new_columns), last_modified_time_);
299 }
300
Empty() const301 bool RowValue::Empty() const {
302 return columns_.empty();
303 }
304
Deserialize(const char * src,std::size_t size)305 RowValue RowValue::Deserialize(const char *src, std::size_t size) {
306 std::size_t offset = 0;
307 assert(size >= sizeof(local_deletion_time_) + sizeof(marked_for_delete_at_));
308 int32_t local_deletion_time =
309 ROCKSDB_NAMESPACE::cassandra::Deserialize<int32_t>(src, offset);
310 offset += sizeof(int32_t);
311 int64_t marked_for_delete_at =
312 ROCKSDB_NAMESPACE::cassandra::Deserialize<int64_t>(src, offset);
313 offset += sizeof(int64_t);
314 if (offset == size) {
315 return RowValue(local_deletion_time, marked_for_delete_at);
316 }
317
318 assert(local_deletion_time == kDefaultLocalDeletionTime);
319 assert(marked_for_delete_at == kDefaultMarkedForDeleteAt);
320 Columns columns;
321 int64_t last_modified_time = 0;
322 while (offset < size) {
323 auto c = ColumnBase::Deserialize(src, offset);
324 offset += c -> Size();
325 assert(offset <= size);
326 last_modified_time = std::max(last_modified_time, c -> Timestamp());
327 columns.push_back(std::move(c));
328 }
329
330 return RowValue(std::move(columns), last_modified_time);
331 }
332
333 // Merge multiple row values into one.
334 // For each column in rows with same index, we pick the one with latest
335 // timestamp. And we also take row tombstone into consideration, by iterating
336 // each row from reverse timestamp order, and stop once we hit the first
337 // row tombstone.
Merge(std::vector<RowValue> && values)338 RowValue RowValue::Merge(std::vector<RowValue>&& values) {
339 assert(values.size() > 0);
340 if (values.size() == 1) {
341 return std::move(values[0]);
342 }
343
344 // Merge columns by their last modified time, and skip once we hit
345 // a row tombstone.
346 std::sort(values.begin(), values.end(),
347 [](const RowValue& r1, const RowValue& r2) {
348 return r1.LastModifiedTime() > r2.LastModifiedTime();
349 });
350
351 std::map<int8_t, std::shared_ptr<ColumnBase>> merged_columns;
352 int64_t tombstone_timestamp = 0;
353
354 for (auto& value : values) {
355 if (value.IsTombstone()) {
356 if (merged_columns.size() == 0) {
357 return std::move(value);
358 }
359 tombstone_timestamp = value.LastModifiedTime();
360 break;
361 }
362 for (auto& column : value.columns_) {
363 int8_t index = column->Index();
364 if (merged_columns.find(index) == merged_columns.end()) {
365 merged_columns[index] = column;
366 } else {
367 if (column->Timestamp() > merged_columns[index]->Timestamp()) {
368 merged_columns[index] = column;
369 }
370 }
371 }
372 }
373
374 int64_t last_modified_time = 0;
375 Columns columns;
376 for (auto& pair: merged_columns) {
377 // For some row, its last_modified_time > row tombstone_timestamp, but
378 // it might have rows whose timestamp is ealier than tombstone, so we
379 // ned to filter these rows.
380 if (pair.second->Timestamp() <= tombstone_timestamp) {
381 continue;
382 }
383 last_modified_time = std::max(last_modified_time, pair.second->Timestamp());
384 columns.push_back(std::move(pair.second));
385 }
386 return RowValue(std::move(columns), last_modified_time);
387 }
388
389 } // namepsace cassandrda
390 } // namespace ROCKSDB_NAMESPACE
391