diff --git a/db/db_impl.cc b/db/db_impl.cc index b18341947..00b5b0a52 100644 --- a/db/db_impl.cc +++ b/db/db_impl.cc @@ -2827,6 +2827,7 @@ Status DBImpl::DoCompactionWork(CompactionState* compact, compact->CleanupBatchBuffer(); compact->CleanupMergedBuffer(); compact->cur_prefix_ = kNullString; + bool prefix_initialized = false; int64_t imm_micros = 0; // Micros spent doing imm_ compactions ColumnFamilyData* cfd = compact->compaction->column_family_data(); @@ -2903,8 +2904,9 @@ Status DBImpl::DoCompactionWork(CompactionState* compact, const SliceTransform* transformer = cfd->options()->compaction_filter_factory_v2->GetPrefixExtractor(); std::string key_prefix = transformer->Transform(key).ToString(); - if (compact->cur_prefix_ == kNullString) { + if (!prefix_initialized) { compact->cur_prefix_ = key_prefix; + prefix_initialized = true; } if (!ParseInternalKey(key, &ikey)) { // log error diff --git a/db/db_test.cc b/db/db_test.cc index 57adc6857..3a5129898 100644 --- a/db/db_test.cc +++ b/db/db_test.cc @@ -3969,6 +3969,56 @@ TEST(DBTest, CompactionFilterV2WithValueChange) { } } +TEST(DBTest, CompactionFilterV2NULLPrefix) { + Options options = CurrentOptions(); + options.num_levels = 3; + options.max_mem_compaction_level = 0; + auto prefix_extractor = NewFixedPrefixTransform(8); + options.compaction_filter_factory_v2 = + std::make_shared(prefix_extractor); + // In a testing environment, we can only flush the application + // compaction filter buffer using universal compaction + option_config_ = kUniversalCompaction; + options.compaction_style = (rocksdb::CompactionStyle)1; + Reopen(&options); + + // Write 100K+1 keys, these are written to a few files + // in L0. We do this so that the current snapshot points + // to the 100001 key.The compaction filter is not invoked + // on keys that are visible via a snapshot because we + // anyways cannot delete it. + const std::string value(10, 'x'); + char first_key[100]; + snprintf(first_key, sizeof(first_key), "%s0000%010d", "NULL", 1); + Put(first_key, value); + for (int i = 1; i < 100000; i++) { + char key[100]; + snprintf(key, sizeof(key), "%08d%010d", i, i); + Put(key, value); + } + + char last_key[100]; + snprintf(last_key, sizeof(last_key), "%s0000%010d", "NULL", 2); + Put(last_key, value); + + // push all files to lower levels + dbfull()->TEST_FlushMemTable(); + dbfull()->TEST_CompactRange(0, nullptr, nullptr); + + // verify that all keys now have the new value that + // was set by the compaction process. + std::string newvalue = Get(first_key); + ASSERT_EQ(newvalue.compare(NEW_VALUE), 0); + newvalue = Get(last_key); + ASSERT_EQ(newvalue.compare(NEW_VALUE), 0); + for (int i = 1; i < 100000; i++) { + char key[100]; + snprintf(key, sizeof(key), "%08d%010d", i, i); + std::string newvalue = Get(key); + ASSERT_EQ(newvalue.compare(NEW_VALUE), 0); + } +} + TEST(DBTest, SparseMerge) { do { Options options = CurrentOptions(); diff --git a/db/merge_operator.cc b/db/merge_operator.cc index 43a8df371..a01d389e9 100644 --- a/db/merge_operator.cc +++ b/db/merge_operator.cc @@ -18,10 +18,13 @@ bool MergeOperator::PartialMergeMulti(const Slice& key, const std::deque& operand_list, std::string* new_value, Logger* logger) const { + assert(operand_list.size() >= 2); // Simply loop through the operands std::string temp_value; - Slice temp_slice; - for (const auto& operand : operand_list) { + Slice temp_slice(operand_list[0]); + + for (int i = 1; i < operand_list.size(); ++i) { + auto& operand = operand_list[i]; if (!PartialMerge(key, temp_slice, operand, &temp_value, logger)) { return false; } diff --git a/util/env_posix.cc b/util/env_posix.cc index 856d49250..237038fcb 100644 --- a/util/env_posix.cc +++ b/util/env_posix.cc @@ -1363,7 +1363,7 @@ class PosixEnv : public Env { EnvOptions OptimizeForLogWrite(const EnvOptions& env_options) const { EnvOptions optimized = env_options; optimized.use_mmap_writes = false; - optimized.fallocate_with_keep_size = true; + optimized.fallocate_with_keep_size = false; return optimized; }