Skip to content

Commit 33ea11e

Browse files
royguoroyguo
committed
[WIP] LinkCompaction: Add LinkSstIterator
1 parent c9dd2ff commit 33ea11e

9 files changed

Lines changed: 578 additions & 54 deletions

File tree

‎db/compaction_job.cc‎

Lines changed: 3 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -1585,34 +1585,8 @@ void CompactionJob::ProcessKeyValueCompaction(SubcompactionState* sub_compact) {
15851585
snapshot_checker_, compact_->compaction->level(),
15861586
db_options_.statistics.get(), shutting_down_);
15871587

1588-
struct BuilderSeparateHelper : public SeparateHelper {
1589-
SeparateHelper* separate_helper = nullptr;
1590-
std::unique_ptr<ValueExtractor> value_meta_extractor;
1591-
Status (*trans_to_separate_callback)(void* args, const Slice& key,
1592-
LazyBuffer& value) = nullptr;
1593-
void* trans_to_separate_callback_args = nullptr;
1594-
1595-
Status TransToSeparate(const Slice& internal_key, LazyBuffer& value,
1596-
const Slice& meta, bool is_merge,
1597-
bool is_index) override {
1598-
return SeparateHelper::TransToSeparate(
1599-
internal_key, value, value.file_number(), meta, is_merge, is_index,
1600-
value_meta_extractor.get());
1601-
}
1602-
1603-
Status TransToSeparate(const Slice& key, LazyBuffer& value) override {
1604-
if (trans_to_separate_callback == nullptr) {
1605-
return Status::NotSupported();
1606-
}
1607-
return trans_to_separate_callback(trans_to_separate_callback_args, key,
1608-
value);
1609-
}
1588+
BuilderSeparateHelper separate_helper;
16101589

1611-
LazyBuffer TransToCombined(const Slice& user_key, uint64_t sequence,
1612-
const LazyBuffer& value) const override {
1613-
return separate_helper->TransToCombined(user_key, sequence, value);
1614-
}
1615-
} separate_helper;
16161590
if (compact_->compaction->immutable_cf_options()
16171591
->value_meta_extractor_factory != nullptr) {
16181592
ValueExtractorContext context = {cfd->GetID()};
@@ -2048,7 +2022,8 @@ void CompactionJob::ProcessKeyValueCompaction(SubcompactionState* sub_compact) {
20482022

20492023
// TODO(guokuankuan@bytedance.com)
20502024
void CompactionJob::ProcessLinkCompaction(SubcompactionState* sub_compact) {
2051-
return;
2025+
assert(sub_compact != nullptr);
2026+
return ProcessKeyValueCompaction(sub_compact);
20522027
}
20532028

20542029
void CompactionJob::ProcessGarbageCollection(SubcompactionState* sub_compact) {

‎db/compaction_job.h‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,38 @@ class VersionEdit;
6464
class VersionSet;
6565

6666
class CompactionJob {
67+
public:
68+
69+
class BuilderSeparateHelper : public SeparateHelper {
70+
public:
71+
SeparateHelper* separate_helper = nullptr;
72+
std::unique_ptr<ValueExtractor> value_meta_extractor;
73+
Status (*trans_to_separate_callback)(void* args, const Slice& key,
74+
LazyBuffer& value) = nullptr;
75+
void* trans_to_separate_callback_args = nullptr;
76+
77+
Status TransToSeparate(const Slice& internal_key, LazyBuffer& value,
78+
const Slice& meta, bool is_merge,
79+
bool is_index) override {
80+
return SeparateHelper::TransToSeparate(
81+
internal_key, value, value.file_number(), meta, is_merge, is_index,
82+
value_meta_extractor.get());
83+
}
84+
85+
Status TransToSeparate(const Slice& key, LazyBuffer& value) override {
86+
if (trans_to_separate_callback == nullptr) {
87+
return Status::NotSupported();
88+
}
89+
return trans_to_separate_callback(trans_to_separate_callback_args, key,
90+
value);
91+
}
92+
93+
LazyBuffer TransToCombined(const Slice& user_key, uint64_t sequence,
94+
const LazyBuffer& value) const override {
95+
return separate_helper->TransToCombined(user_key, sequence, value);
96+
}
97+
};
98+
6799
public:
68100
CompactionJob(int job_id, Compaction* compaction,
69101
const ImmutableDBOptions& db_options,

‎db/table_cache.cc‎

Lines changed: 34 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -85,7 +85,7 @@ static bool InheritanceMismatch(const FileMetaData& sst_meta,
8585
return true;
8686
}
8787

88-
// Store params for create depend table iterator in future
88+
// Store params for create dependency table iterator in future
8989
class LazyCreateIterator : public Snapshot {
9090
TableCache* table_cache_;
9191
ReadOptions options_; // deep copy
@@ -292,7 +292,7 @@ InternalIterator* TableCache::NewIterator(
292292
}
293293
size_t readahead = 0;
294294
bool record_stats = !for_compaction;
295-
if (file_meta.prop.is_map_sst()) {
295+
if (file_meta.prop.is_map_sst() || file_meta.prop.is_link_sst()) {
296296
record_stats = false;
297297
} else {
298298
// MapSST don't handle these
@@ -340,22 +340,17 @@ InternalIterator* TableCache::NewIterator(
340340
}
341341
InternalIterator* result = nullptr;
342342
if (s.ok()) {
343-
if (!file_meta.prop.is_map_sst()) {
344-
if (options.table_filter &&
345-
!options.table_filter(*table_reader->GetTableProperties())) {
346-
result = NewEmptyInternalIterator<LazyBuffer>(arena);
347-
} else {
348-
result = table_reader->NewIterator(options, prefix_extractor, arena,
349-
skip_filters, for_compaction);
350-
}
351-
} else {
343+
// For map & linked SST, we should expand their underlying key value pairs,
344+
// not simply iterate the input SST key values.
345+
if(file_meta.prop.is_map_sst() || file_meta.prop.is_link_sst()) {
352346
ReadOptions map_options = options;
353347
map_options.total_order_seek = true;
354348
map_options.readahead_size = 0;
355349
result =
356350
table_reader->NewIterator(map_options, prefix_extractor, arena,
357351
skip_filters, false /* for_compaction */);
358352
if (!dependence_map.empty()) {
353+
// Map SST will handle range deletion internally, so we can skip here.
359354
bool ignore_range_deletions =
360355
options.ignore_range_deletions ||
361356
file_meta.prop.map_handle_range_deletions();
@@ -365,33 +360,54 @@ InternalIterator* TableCache::NewIterator(
365360
lazy_create_iter = new (buffer) LazyCreateIterator(
366361
this, options, env_options, range_del_agg, prefix_extractor,
367362
for_compaction, skip_filters, ignore_range_deletions, level);
368-
369363
} else {
370364
lazy_create_iter = new LazyCreateIterator(
371365
this, options, env_options, range_del_agg, prefix_extractor,
372366
for_compaction, skip_filters, ignore_range_deletions, level);
373367
}
374-
auto map_sst_iter = NewMapSstIterator(
375-
&file_meta, result, dependence_map, ioptions_.internal_comparator,
376-
lazy_create_iter, c_style_callback(*lazy_create_iter), arena);
368+
369+
// For map & linked sst, we should expand their dependencies and merge
370+
// all related iterators into one combined iterator for further reads.
371+
InternalIterator* sst_iter = nullptr;
372+
373+
if(file_meta.prop.is_map_sst()) {
374+
sst_iter = NewMapSstIterator(
375+
&file_meta, result, dependence_map, ioptions_.internal_comparator,
376+
lazy_create_iter, c_style_callback(*lazy_create_iter), arena);
377+
} else {
378+
assert(file_meta.prop.is_link_sst());
379+
sst_iter = NewLinkSstIterator(
380+
&file_meta, result, dependence_map, ioptions_.internal_comparator,
381+
lazy_create_iter, c_style_callback(*lazy_create_iter), arena);
382+
}
383+
377384
if (arena != nullptr) {
378-
map_sst_iter->RegisterCleanup(
385+
sst_iter->RegisterCleanup(
379386
[](void* arg1, void* arg2) {
380387
static_cast<InternalIterator*>(arg1)->~InternalIterator();
381388
static_cast<LazyCreateIterator*>(arg2)->~LazyCreateIterator();
382389
},
383390
result, lazy_create_iter);
384391
} else {
385-
map_sst_iter->RegisterCleanup(
392+
sst_iter->RegisterCleanup(
386393
[](void* arg1, void* arg2) {
387394
delete static_cast<InternalIterator*>(arg1);
388395
delete static_cast<LazyCreateIterator*>(arg2);
389396
},
390397
result, lazy_create_iter);
391398
}
392-
result = map_sst_iter;
399+
result = sst_iter;
400+
}
401+
} else {
402+
if (options.table_filter &&
403+
!options.table_filter(*table_reader->GetTableProperties())) {
404+
result = NewEmptyInternalIterator<LazyBuffer>(arena);
405+
} else {
406+
result = table_reader->NewIterator(options, prefix_extractor, arena,
407+
skip_filters, for_compaction);
393408
}
394409
}
410+
395411
if (create_new_table_reader) {
396412
assert(handle == nullptr);
397413
result->RegisterCleanup(&DeleteTableReader, table_reader,

‎db/version_edit.h‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -98,8 +98,11 @@ struct TablePropertyCache {
9898
std::vector<uint64_t> inheritance; // inheritance set
9999
uint64_t earliest_time_begin_compact = port::kMaxUint64;
100100
uint64_t latest_time_end_compact = port::kMaxUint64;
101+
uint32_t lbr_hash_bits = 11; // hash bits for each LinkSST KV
102+
uint32_t lbr_group_size = 16; // Group size for LinkBlockRecord
101103

102104
bool is_map_sst() const { return purpose == kMapSst; }
105+
bool is_link_sst() const {return purpose == kLinkSst; }
103106
bool has_range_deletions() const { return (flags & kNoRangeDeletions) == 0; }
104107
bool map_handle_range_deletions() const {
105108
return (flags & kMapHandleRangeDeletions) != 0;

‎include/rocksdb/options.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,7 @@ enum SstPurpose {
8888
kEssenceSst, // Actual data storage sst
8989
kLogSst, // Log as sst
9090
kMapSst, // Dummy sst
91+
kLinkSst, // Link SST
9192
};
9293

9394
struct Options;

0 commit comments

Comments
 (0)