Separate compaction filter for each compaction

Summary:
If we have same compaction filter for each compaction,
application cannot know about the different compaction processes.
Later on, we can put in more details in compaction filter for the
application to consume and use it according to its needs. For e.g. In
the universal compaction, we have a compaction process involving all the
files while others don't involve all the files. Applications may want to
collect some stats only when during full compaction.

Test Plan: run existing unit tests

Reviewers: haobo, dhruba

Reviewed By: dhruba

CC: xinyaohu, leveldb

Differential Revision: https://reviews.facebook.net/D12057
main
sumeet 11 years ago
parent 9f6b8f0032
commit 3b81df34bd
  1. 15
      db/db_impl.cc
  2. 51
      db/db_test.cc
  3. 25
      include/leveldb/compaction_filter.h
  4. 11
      include/leveldb/options.h
  5. 8
      util/options.cc
  6. 11
      utilities/ttl/db_ttl.cc
  7. 38
      utilities/ttl/db_ttl.h
  8. 29
      utilities/ttl/ttl_test.cc

@ -158,6 +158,10 @@ Options SanitizeOptions(const std::string& dbname,
if (result.soft_rate_limit > result.hard_rate_limit) { if (result.soft_rate_limit > result.hard_rate_limit) {
result.soft_rate_limit = result.hard_rate_limit; result.soft_rate_limit = result.hard_rate_limit;
} }
if (result.compaction_filter &&
result.compaction_filter_factory->CreateCompactionFilter().get()) {
Log(result.info_log, "Both filter and factory specified. Using filter");
}
return result; return result;
} }
@ -1784,6 +1788,13 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) {
MergeHelper merge(user_comparator(), options_.merge_operator, MergeHelper merge(user_comparator(), options_.merge_operator,
options_.info_log.get(), options_.info_log.get(),
false /* internal key corruption is expected */); false /* internal key corruption is expected */);
auto compaction_filter = options_.compaction_filter;
std::unique_ptr<CompactionFilter> compaction_filter_from_factory = nullptr;
if (!compaction_filter) {
compaction_filter_from_factory = std::move(
options_.compaction_filter_factory->CreateCompactionFilter());
compaction_filter = compaction_filter_from_factory.get();
}
for (; input->Valid() && !shutting_down_.Acquire_Load(); ) { for (; input->Valid() && !shutting_down_.Acquire_Load(); ) {
// Prioritize immutable compaction work // Prioritize immutable compaction work
if (imm_.imm_flush_needed.NoBarrier_Load() != nullptr) { if (imm_.imm_flush_needed.NoBarrier_Load() != nullptr) {
@ -1830,7 +1841,7 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) {
visible_in_snapshot = kMaxSequenceNumber; visible_in_snapshot = kMaxSequenceNumber;
// apply the compaction filter to the first occurrence of the user key // apply the compaction filter to the first occurrence of the user key
if (options_.compaction_filter && if (compaction_filter &&
ikey.type == kTypeValue && ikey.type == kTypeValue &&
(visible_at_tip || ikey.sequence > latest_snapshot)) { (visible_at_tip || ikey.sequence > latest_snapshot)) {
// If the user has specified a compaction filter and the sequence // If the user has specified a compaction filter and the sequence
@ -1841,7 +1852,7 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) {
bool value_changed = false; bool value_changed = false;
compaction_filter_value.clear(); compaction_filter_value.clear();
bool to_delete = bool to_delete =
options_.compaction_filter->Filter(compact->compaction->level(), compaction_filter->Filter(compact->compaction->level(),
ikey.user_key, value, ikey.user_key, value,
&compaction_filter_value, &compaction_filter_value,
&value_changed); &value_changed);

@ -1698,12 +1698,52 @@ class ChangeFilter : public CompactionFilter {
const int argv_; const int argv_;
}; };
class KeepFilterFactory : public CompactionFilterFactory {
public:
virtual std::unique_ptr<CompactionFilter>
CreateCompactionFilter() override {
return std::unique_ptr<CompactionFilter>(new KeepFilter());
}
virtual const char* Name() const override {
return "KeepFilterFactory";
}
};
class DeleteFilterFactory : public CompactionFilterFactory {
public:
virtual std::unique_ptr<CompactionFilter>
CreateCompactionFilter() override {
return std::unique_ptr<CompactionFilter>(new DeleteFilter());
}
virtual const char* Name() const override {
return "DeleteFilterFactory";
}
};
class ChangeFilterFactory : public CompactionFilterFactory {
public:
explicit ChangeFilterFactory(int argv) : argv_(argv) {}
virtual std::unique_ptr<CompactionFilter>
CreateCompactionFilter() override {
return std::unique_ptr<CompactionFilter>(new ChangeFilter(argv_));
}
virtual const char* Name() const override {
return "ChangeFilterFactory";
}
private:
const int argv_;
};
TEST(DBTest, CompactionFilter) { TEST(DBTest, CompactionFilter) {
Options options = CurrentOptions(); Options options = CurrentOptions();
options.num_levels = 3; options.num_levels = 3;
options.max_mem_compaction_level = 0; options.max_mem_compaction_level = 0;
auto keep_filter = std::make_shared<KeepFilter>(); options.compaction_filter_factory = std::make_shared<KeepFilterFactory>();
options.compaction_filter = keep_filter.get();
Reopen(&options); Reopen(&options);
// Write 100K keys, these are written to a few files in L0. // Write 100K keys, these are written to a few files in L0.
@ -1778,8 +1818,7 @@ TEST(DBTest, CompactionFilter) {
// create a new database with the compaction // create a new database with the compaction
// filter in such a way that it deletes all keys // filter in such a way that it deletes all keys
auto delete_filter = std::make_shared<DeleteFilter>(); options.compaction_filter_factory = std::make_shared<DeleteFilterFactory>();
options.compaction_filter = delete_filter.get();
options.create_if_missing = true; options.create_if_missing = true;
DestroyAndReopen(&options); DestroyAndReopen(&options);
@ -1843,8 +1882,8 @@ TEST(DBTest, CompactionFilterWithValueChange) {
Options options = CurrentOptions(); Options options = CurrentOptions();
options.num_levels = 3; options.num_levels = 3;
options.max_mem_compaction_level = 0; options.max_mem_compaction_level = 0;
auto change_filter = std::make_shared<ChangeFilter>(100); options.compaction_filter_factory =
options.compaction_filter = change_filter.get(); std::make_shared<ChangeFilterFactory>(100);
Reopen(&options); Reopen(&options);
// Write 100K+1 keys, these are written to a few files // Write 100K+1 keys, these are written to a few files

@ -41,6 +41,31 @@ class CompactionFilter {
virtual const char* Name() const = 0; virtual const char* Name() const = 0;
}; };
// Each compaction will create a new CompactionFilter allowing the
// application to know about different campactions
class CompactionFilterFactory {
public:
virtual ~CompactionFilterFactory() { };
virtual std::unique_ptr<CompactionFilter> CreateCompactionFilter() = 0;
// Returns a name that identifies this compaction filter factory.
virtual const char* Name() const = 0;
};
// Default implementaion of CompactionFilterFactory which does not
// return any filter
class DefaultCompactionFilterFactory : public CompactionFilterFactory {
public:
virtual std::unique_ptr<CompactionFilter>
CreateCompactionFilter() override {
return std::unique_ptr<CompactionFilter>(nullptr);
}
virtual const char* Name() const override {
return "DefaultCompactionFilterFactory";
}
};
} // namespace leveldb } // namespace leveldb
#endif // STORAGE_LEVELDB_INCLUDE_COMPACTION_FILTER_H_ #endif // STORAGE_LEVELDB_INCLUDE_COMPACTION_FILTER_H_

@ -25,6 +25,7 @@ class Logger;
class MergeOperator; class MergeOperator;
class Snapshot; class Snapshot;
class CompactionFilter; class CompactionFilter;
class CompactionFilterFactory;
using std::shared_ptr; using std::shared_ptr;
@ -84,8 +85,13 @@ struct Options {
// Default: nullptr // Default: nullptr
const MergeOperator* merge_operator; const MergeOperator* merge_operator;
// The client must provide compaction_filter_factory if it requires a new
// compaction filter to be used for different compaction processes
// Allows an application to modify/delete a key-value during background // Allows an application to modify/delete a key-value during background
// compaction. // compaction.
// Ideally, client should specify only one of filter or factory.
// compaction_filter takes precedence over compaction_filter_factory if
// client specifies both.
// Default: nullptr // Default: nullptr
const CompactionFilter* compaction_filter; const CompactionFilter* compaction_filter;
@ -506,6 +512,11 @@ struct Options {
// Default: a factory that provides a skip-list-based implementation of // Default: a factory that provides a skip-list-based implementation of
// MemTableRep. // MemTableRep.
std::shared_ptr<MemTableRepFactory> memtable_factory; std::shared_ptr<MemTableRepFactory> memtable_factory;
// This is a factory that provides compaction filter objects which allow
// an application to modify/delete a key-value during background compaction.
// Default: a factory that doesn't provide any object
std::shared_ptr<CompactionFilterFactory> compaction_filter_factory;
}; };
// Options that control read operations // Options that control read operations

@ -80,7 +80,11 @@ Options::Options()
bytes_per_sync(0), bytes_per_sync(0),
compaction_style(kCompactionStyleLevel), compaction_style(kCompactionStyleLevel),
filter_deletes(false), filter_deletes(false),
memtable_factory(std::shared_ptr<SkipListFactory>(new SkipListFactory)) { memtable_factory(std::shared_ptr<SkipListFactory>(new SkipListFactory)),
compaction_filter_factory(
std::shared_ptr<CompactionFilterFactory>(
new DefaultCompactionFilterFactory())) {
assert(memtable_factory.get() != nullptr); assert(memtable_factory.get() != nullptr);
} }
@ -96,6 +100,8 @@ Options::Dump(Logger* log) const
merge_operator? merge_operator->Name() : "None"); merge_operator? merge_operator->Name() : "None");
Log(log," Options.compaction_filter: %s", Log(log," Options.compaction_filter: %s",
compaction_filter? compaction_filter->Name() : "None"); compaction_filter? compaction_filter->Name() : "None");
Log(log," Options.compaction_filter_factory: %s",
compaction_filter_factory->Name());
Log(log," Options.error_if_exists: %d", error_if_exists); Log(log," Options.error_if_exists: %d", error_if_exists);
Log(log," Options.paranoid_checks: %d", paranoid_checks); Log(log," Options.paranoid_checks: %d", paranoid_checks);
Log(log," Options.env: %p", env); Log(log," Options.env: %p", env);

@ -20,9 +20,16 @@ DBWithTTL::DBWithTTL(const int32_t ttl,
ttl_(ttl) { ttl_(ttl) {
Options options_to_open = options; Options options_to_open = options;
ttl_comp_filter_.reset(new TtlCompactionFilter(ttl, if (options.compaction_filter) {
options.compaction_filter)); ttl_comp_filter_.reset(
new TtlCompactionFilter(ttl, options.compaction_filter));
options_to_open.compaction_filter = ttl_comp_filter_.get(); options_to_open.compaction_filter = ttl_comp_filter_.get();
} else {
options_to_open.compaction_filter_factory =
std::shared_ptr<CompactionFilterFactory>(
new TtlCompactionFilterFactory(
ttl, options.compaction_filter_factory));
}
if (options.merge_operator) { if (options.merge_operator) {
ttl_merge_op_.reset(new TtlMergeOperator(options.merge_operator)); ttl_merge_op_.reset(new TtlMergeOperator(options.merge_operator));

@ -176,11 +176,19 @@ class TtlIterator : public Iterator {
class TtlCompactionFilter : public CompactionFilter { class TtlCompactionFilter : public CompactionFilter {
public: public:
TtlCompactionFilter(int32_t ttl, const CompactionFilter* comp_filter) TtlCompactionFilter(
int32_t ttl,
const CompactionFilter* user_comp_filter,
std::unique_ptr<const CompactionFilter>
user_comp_filter_from_factory = nullptr)
: ttl_(ttl), : ttl_(ttl),
user_comp_filter_(comp_filter) { user_comp_filter_(user_comp_filter),
user_comp_filter_from_factory_(std::move(user_comp_filter_from_factory)) {
// Unlike the merge operator, compaction filter is necessary for TTL, hence // Unlike the merge operator, compaction filter is necessary for TTL, hence
// this would be called even if user doesn't specify any compaction-filter // this would be called even if user doesn't specify any compaction-filter
if (!user_comp_filter_) {
user_comp_filter_ = user_comp_filter_from_factory_.get();
}
} }
virtual bool Filter(int level, virtual bool Filter(int level,
@ -215,6 +223,32 @@ class TtlCompactionFilter : public CompactionFilter {
private: private:
int32_t ttl_; int32_t ttl_;
const CompactionFilter* user_comp_filter_; const CompactionFilter* user_comp_filter_;
std::unique_ptr<const CompactionFilter> user_comp_filter_from_factory_;
};
class TtlCompactionFilterFactory : public CompactionFilterFactory {
public:
TtlCompactionFilterFactory(
int32_t ttl,
std::shared_ptr<CompactionFilterFactory> comp_filter_factory)
: ttl_(ttl),
user_comp_filter_factory_(comp_filter_factory) { }
virtual std::unique_ptr<CompactionFilter> CreateCompactionFilter() {
return std::unique_ptr<TtlCompactionFilter>(
new TtlCompactionFilter(
ttl_,
nullptr,
std::move(user_comp_filter_factory_->CreateCompactionFilter())));
}
virtual const char* Name() const override {
return "TtlCompactionFilterFactory";
}
private:
int32_t ttl_;
std::shared_ptr<CompactionFilterFactory> user_comp_filter_factory_;
}; };
class TtlMergeOperator : public MergeOperator { class TtlMergeOperator : public MergeOperator {

@ -2,6 +2,7 @@
// Use of this source code is governed by a BSD-style license that can be // Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file. See the AUTHORS file for names of contributors. // found in the LICENSE file. See the AUTHORS file for names of contributors.
#include <memory>
#include "leveldb/compaction_filter.h" #include "leveldb/compaction_filter.h"
#include "utilities/utility_db.h" #include "utilities/utility_db.h"
#include "util/testharness.h" #include "util/testharness.h"
@ -54,8 +55,9 @@ class TtlTest {
// Open with TestFilter compaction filter // Open with TestFilter compaction filter
void OpenTtlWithTestCompaction(int32_t ttl) { void OpenTtlWithTestCompaction(int32_t ttl) {
test_comp_filter_.reset(new TestFilter(kSampleSize_, kNewValue_)); options_.compaction_filter_factory =
options_.compaction_filter = test_comp_filter_.get(); std::shared_ptr<CompactionFilterFactory>(
new TestFilterFactory(kSampleSize_, kNewValue_));
OpenTtl(ttl); OpenTtl(ttl);
} }
@ -252,6 +254,29 @@ class TtlTest {
const std::string kNewValue_; const std::string kNewValue_;
}; };
class TestFilterFactory : public CompactionFilterFactory {
public:
TestFilterFactory(const int64_t kSampleSize, const std::string kNewValue)
: kSampleSize_(kSampleSize),
kNewValue_(kNewValue) {
}
virtual std::unique_ptr<CompactionFilter>
CreateCompactionFilter() override {
return std::unique_ptr<CompactionFilter>(
new TestFilter(kSampleSize_, kNewValue_));
}
virtual const char* Name() const override {
return "TestFilterFactory";
}
private:
const int64_t kSampleSize_;
const std::string kNewValue_;
};
// Choose carefully so that Put, Gets & Compaction complete in 1 second buffer // Choose carefully so that Put, Gets & Compaction complete in 1 second buffer
const int64_t kSampleSize_ = 100; const int64_t kSampleSize_ = 100;

Loading…
Cancel
Save