class KeepFilter : public CompactionFilter {
public:
- virtual bool Filter(int /*level*/, const Slice& /*key*/,
- const Slice& /*value*/, std::string* /*new_value*/,
- bool* /*value_changed*/) const override {
+ bool Filter(int /*level*/, const Slice& /*key*/, const Slice& /*value*/,
+ std::string* /*new_value*/,
+ bool* /*value_changed*/) const override {
cfilter_count++;
return false;
}
- virtual const char* Name() const override { return "KeepFilter"; }
+ const char* Name() const override { return "KeepFilter"; }
};
class DeleteFilter : public CompactionFilter {
public:
- virtual bool Filter(int /*level*/, const Slice& /*key*/,
- const Slice& /*value*/, std::string* /*new_value*/,
- bool* /*value_changed*/) const override {
+ bool Filter(int /*level*/, const Slice& /*key*/, const Slice& /*value*/,
+ std::string* /*new_value*/,
+ bool* /*value_changed*/) const override {
cfilter_count++;
return true;
}
- virtual const char* Name() const override { return "DeleteFilter"; }
+ const char* Name() const override { return "DeleteFilter"; }
};
class DeleteISFilter : public CompactionFilter {
public:
- virtual bool Filter(int /*level*/, const Slice& key, const Slice& /*value*/,
- std::string* /*new_value*/,
- bool* /*value_changed*/) const override {
+ bool Filter(int /*level*/, const Slice& key, const Slice& /*value*/,
+ std::string* /*new_value*/,
+ bool* /*value_changed*/) const override {
cfilter_count++;
int i = std::stoi(key.ToString());
if (i > 5 && i <= 105) {
return false;
}
- virtual bool IgnoreSnapshots() const override { return true; }
+ bool IgnoreSnapshots() const override { return true; }
- virtual const char* Name() const override { return "DeleteFilter"; }
+ const char* Name() const override { return "DeleteFilter"; }
};
// Skip x if floor(x/10) is even, use range skips. Requires that keys are
// zero-padded to length 10.
class SkipEvenFilter : public CompactionFilter {
public:
- virtual Decision FilterV2(int /*level*/, const Slice& key,
- ValueType /*value_type*/,
- const Slice& /*existing_value*/,
- std::string* /*new_value*/,
- std::string* skip_until) const override {
+ Decision FilterV2(int /*level*/, const Slice& key, ValueType /*value_type*/,
+ const Slice& /*existing_value*/, std::string* /*new_value*/,
+ std::string* skip_until) const override {
cfilter_count++;
int i = std::stoi(key.ToString());
if (i / 10 % 2 == 0) {
return Decision::kKeep;
}
- virtual bool IgnoreSnapshots() const override { return true; }
+ bool IgnoreSnapshots() const override { return true; }
- virtual const char* Name() const override { return "DeleteFilter"; }
+ const char* Name() const override { return "DeleteFilter"; }
};
class DelayFilter : public CompactionFilter {
public:
explicit DelayFilter(DBTestBase* d) : db_test(d) {}
- virtual bool Filter(int /*level*/, const Slice& /*key*/,
- const Slice& /*value*/, std::string* /*new_value*/,
- bool* /*value_changed*/) const override {
+ bool Filter(int /*level*/, const Slice& /*key*/, const Slice& /*value*/,
+ std::string* /*new_value*/,
+ bool* /*value_changed*/) const override {
db_test->env_->addon_time_.fetch_add(1000);
return true;
}
- virtual const char* Name() const override { return "DelayFilter"; }
+ const char* Name() const override { return "DelayFilter"; }
private:
DBTestBase* db_test;
public:
explicit ConditionalFilter(const std::string* filtered_value)
: filtered_value_(filtered_value) {}
- virtual bool Filter(int /*level*/, const Slice& /*key*/, const Slice& value,
- std::string* /*new_value*/,
- bool* /*value_changed*/) const override {
+ bool Filter(int /*level*/, const Slice& /*key*/, const Slice& value,
+ std::string* /*new_value*/,
+ bool* /*value_changed*/) const override {
return value.ToString() == *filtered_value_;
}
- virtual const char* Name() const override { return "ConditionalFilter"; }
+ const char* Name() const override { return "ConditionalFilter"; }
private:
const std::string* filtered_value_;
public:
explicit ChangeFilter() {}
- virtual bool Filter(int /*level*/, const Slice& /*key*/,
- const Slice& /*value*/, std::string* new_value,
- bool* value_changed) const override {
+ bool Filter(int /*level*/, const Slice& /*key*/, const Slice& /*value*/,
+ std::string* new_value, bool* value_changed) const override {
assert(new_value != nullptr);
*new_value = NEW_VALUE;
*value_changed = true;
return false;
}
- virtual const char* Name() const override { return "ChangeFilter"; }
+ const char* Name() const override { return "ChangeFilter"; }
};
class KeepFilterFactory : public CompactionFilterFactory {
check_context_cf_id_(check_context_cf_id),
compaction_filter_created_(false) {}
- virtual std::unique_ptr<CompactionFilter> CreateCompactionFilter(
+ std::unique_ptr<CompactionFilter> CreateCompactionFilter(
const CompactionFilter::Context& context) override {
if (check_context_) {
EXPECT_EQ(expect_full_compaction_.load(), context.is_full_compaction);
bool compaction_filter_created() const { return compaction_filter_created_; }
- virtual const char* Name() const override { return "KeepFilterFactory"; }
+ const char* Name() const override { return "KeepFilterFactory"; }
bool check_context_;
bool check_context_cf_id_;
std::atomic_bool expect_full_compaction_;
class DeleteFilterFactory : public CompactionFilterFactory {
public:
- virtual std::unique_ptr<CompactionFilter> CreateCompactionFilter(
+ std::unique_ptr<CompactionFilter> CreateCompactionFilter(
const CompactionFilter::Context& context) override {
if (context.is_manual_compaction) {
return std::unique_ptr<CompactionFilter>(new DeleteFilter());
}
}
- virtual const char* Name() const override { return "DeleteFilterFactory"; }
+ const char* Name() const override { return "DeleteFilterFactory"; }
};
// Delete Filter Factory which ignores snapshots
class DeleteISFilterFactory : public CompactionFilterFactory {
public:
- virtual std::unique_ptr<CompactionFilter> CreateCompactionFilter(
+ std::unique_ptr<CompactionFilter> CreateCompactionFilter(
const CompactionFilter::Context& context) override {
if (context.is_manual_compaction) {
return std::unique_ptr<CompactionFilter>(new DeleteISFilter());
}
}
- virtual const char* Name() const override { return "DeleteFilterFactory"; }
+ const char* Name() const override { return "DeleteFilterFactory"; }
};
class SkipEvenFilterFactory : public CompactionFilterFactory {
public:
- virtual std::unique_ptr<CompactionFilter> CreateCompactionFilter(
+ std::unique_ptr<CompactionFilter> CreateCompactionFilter(
const CompactionFilter::Context& context) override {
if (context.is_manual_compaction) {
return std::unique_ptr<CompactionFilter>(new SkipEvenFilter());
}
}
- virtual const char* Name() const override { return "SkipEvenFilterFactory"; }
+ const char* Name() const override { return "SkipEvenFilterFactory"; }
};
class DelayFilterFactory : public CompactionFilterFactory {
public:
explicit DelayFilterFactory(DBTestBase* d) : db_test(d) {}
- virtual std::unique_ptr<CompactionFilter> CreateCompactionFilter(
+ std::unique_ptr<CompactionFilter> CreateCompactionFilter(
const CompactionFilter::Context& /*context*/) override {
return std::unique_ptr<CompactionFilter>(new DelayFilter(db_test));
}
- virtual const char* Name() const override { return "DelayFilterFactory"; }
+ const char* Name() const override { return "DelayFilterFactory"; }
private:
DBTestBase* db_test;
explicit ConditionalFilterFactory(const Slice& filtered_value)
: filtered_value_(filtered_value.ToString()) {}
- virtual std::unique_ptr<CompactionFilter> CreateCompactionFilter(
+ std::unique_ptr<CompactionFilter> CreateCompactionFilter(
const CompactionFilter::Context& /*context*/) override {
return std::unique_ptr<CompactionFilter>(
new ConditionalFilter(&filtered_value_));
}
- virtual const char* Name() const override {
- return "ConditionalFilterFactory";
- }
+ const char* Name() const override { return "ConditionalFilterFactory"; }
private:
std::string filtered_value_;
public:
explicit ChangeFilterFactory() {}
- virtual std::unique_ptr<CompactionFilter> CreateCompactionFilter(
+ std::unique_ptr<CompactionFilter> CreateCompactionFilter(
const CompactionFilter::Context& /*context*/) override {
return std::unique_ptr<CompactionFilter>(new ChangeFilter());
}
- virtual const char* Name() const override { return "ChangeFilterFactory"; }
+ const char* Name() const override { return "ChangeFilterFactory"; }
};
#ifndef ROCKSDB_LITE
Arena arena;
{
InternalKeyComparator icmp(options.comparator);
- RangeDelAggregator range_del_agg(icmp, {} /* snapshots */);
- ScopedArenaIterator iter(
- dbfull()->NewInternalIterator(&arena, &range_del_agg, handles_[1]));
+ ReadRangeDelAggregator range_del_agg(&icmp,
+ kMaxSequenceNumber /* upper_bound */);
+ ScopedArenaIterator iter(dbfull()->NewInternalIterator(
+ &arena, &range_del_agg, kMaxSequenceNumber, handles_[1]));
iter->SeekToFirst();
ASSERT_OK(iter->status());
while (iter->Valid()) {
}
}
ASSERT_EQ(total, 100000);
- ASSERT_EQ(count, 1);
+ ASSERT_EQ(count, 0);
// overwrite all the 100K keys once again.
for (int i = 0; i < 100000; i++) {
count = 0;
{
InternalKeyComparator icmp(options.comparator);
- RangeDelAggregator range_del_agg(icmp, {} /* snapshots */);
- ScopedArenaIterator iter(
- dbfull()->NewInternalIterator(&arena, &range_del_agg, handles_[1]));
+ ReadRangeDelAggregator range_del_agg(&icmp,
+ kMaxSequenceNumber /* upper_bound */);
+ ScopedArenaIterator iter(dbfull()->NewInternalIterator(
+ &arena, &range_del_agg, kMaxSequenceNumber, handles_[1]));
iter->SeekToFirst();
ASSERT_OK(iter->status());
while (iter->Valid()) {
int total = 0;
Arena arena;
InternalKeyComparator icmp(options.comparator);
- RangeDelAggregator range_del_agg(icmp, {} /* snapshots */);
- ScopedArenaIterator iter(
- dbfull()->NewInternalIterator(&arena, &range_del_agg));
+ ReadRangeDelAggregator range_del_agg(&icmp,
+ kMaxSequenceNumber /* snapshots */);
+ ScopedArenaIterator iter(dbfull()->NewInternalIterator(
+ &arena, &range_del_agg, kMaxSequenceNumber));
iter->SeekToFirst();
ASSERT_OK(iter->status());
while (iter->Valid()) {
iter->Next();
}
ASSERT_EQ(total, 700);
- ASSERT_EQ(count, 1);
+ ASSERT_EQ(count, 0);
}
}
#endif // ROCKSDB_LITE
}
#ifndef ROCKSDB_LITE
-// Compaction filters should only be applied to records that are newer than the
-// latest snapshot. This test inserts records and applies a delete filter.
-TEST_F(DBTestCompactionFilter, CompactionFilterSnapshot) {
- Options options = CurrentOptions();
- options.compaction_filter_factory = std::make_shared<DeleteFilterFactory>();
- options.disable_auto_compactions = true;
- options.create_if_missing = true;
- DestroyAndReopen(options);
-
- // Put some data.
- const Snapshot* snapshot = nullptr;
- for (int table = 0; table < 4; ++table) {
- for (int i = 0; i < 10; ++i) {
- Put(ToString(table * 100 + i), "val");
- }
- Flush();
-
- if (table == 0) {
- snapshot = db_->GetSnapshot();
- }
- }
- assert(snapshot != nullptr);
-
- cfilter_count = 0;
- ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
- // The filter should delete 10 records.
- ASSERT_EQ(30U, cfilter_count);
-
- // Release the snapshot and compact again -> now all records should be
- // removed.
- db_->ReleaseSnapshot(snapshot);
- ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
- ASSERT_EQ(0U, CountLiveFiles());
-}
-
-// Compaction filters should only be applied to records that are newer than the
-// latest snapshot. However, if the compaction filter asks to ignore snapshots
-// records newer than the snapshot will also be processed
+// Compaction filters aplies to all records, regardless snapshots.
TEST_F(DBTestCompactionFilter, CompactionFilterIgnoreSnapshot) {
std::string five = ToString(5);
Options options = CurrentOptions();
DestroyAndReopen(options);
Put("0000000010", "v10");
- Put("0000000020", "v20"); // skipped
+ Put("0000000020", "v20"); // skipped
Put("0000000050", "v50");
Flush();
EXPECT_EQ("v50", val);
}
+class TestNotSupportedFilter : public CompactionFilter {
+ public:
+ bool Filter(int /*level*/, const Slice& /*key*/, const Slice& /*value*/,
+ std::string* /*new_value*/,
+ bool* /*value_changed*/) const override {
+ return true;
+ }
+
+ const char* Name() const override { return "NotSupported"; }
+ bool IgnoreSnapshots() const override { return false; }
+};
+
+TEST_F(DBTestCompactionFilter, IgnoreSnapshotsFalse) {
+ Options options = CurrentOptions();
+ options.compaction_filter = new TestNotSupportedFilter();
+ DestroyAndReopen(options);
+
+ Put("a", "v10");
+ Put("z", "v20");
+ Flush();
+
+ Put("a", "v10");
+ Put("z", "v20");
+ Flush();
+
+ // Comapction should fail because IgnoreSnapshots() = false
+ EXPECT_TRUE(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr)
+ .IsNotSupported());
+
+ delete options.compaction_filter;
+}
+
} // namespace rocksdb
int main(int argc, char** argv) {
projects / ceph.git / blobdiff