Packages
rocksdb
2.4.1
3.1.1
3.1.0
3.0.0
2.6.2
2.6.1
retired
2.6.0
retired
2.5.0
2.4.1
2.4.0
2.3.0
2.2.0
2.1.0
2.0.0
1.9.0
1.8.0
1.7.0
1.6.0
1.5.1
1.5.0
1.4.0
1.3.2
1.3.1
1.3.0
1.2.0
1.1.1
1.1.0
1.0.0
0.26.2
0.26.1
0.26.0
0.25.0
0.24.0
0.23.3
0.23.2
0.23.1
0.23.0
0.22.0
0.21.0
0.20.1
0.20.0
0.19.0
0.18.0
0.17.0
0.16.0
0.15.0
0.14.0
0.13.1
0.13.0
0.12.0
0.11.0
0.10.0
0.9.1
0.9.0
0.8.2
0.8.1
0.8.0
0.7.1
0.7.0
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
RocksDB for Erlang
Current section
Files
Jump to
Current section
Files
c_src/posting_list_merge_operator.cc
// Copyright (c) 2018-2026 Benoit Chesneau
//
// This file is provided to you under the Apache License,
// Version 2.0 (the "License"); you may not use this file
// except in compliance with the License. You may obtain
// a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
#include <memory>
#include <deque>
#include <string>
#include <cstdint>
#include <unordered_map>
#include "rocksdb/slice.h"
#include "rocksdb/merge_operator.h"
#include "erl_nif.h"
#include "atoms.h"
#include "posting_list_merge_operator.h"
namespace erocksdb {
PostingListMergeOperator::PostingListMergeOperator() {}
// Parse an existing posting list binary into a map of key -> is_tombstone
void PostingListMergeOperator::ParseExistingValue(
const rocksdb::Slice& value,
std::unordered_map<std::string, bool>& key_states) const {
const char* ptr = value.data();
const char* end = ptr + value.size();
while (ptr + 5 <= end) { // 4 bytes length + 1 byte flag minimum
uint32_t len = (static_cast<uint8_t>(ptr[0]) << 24) |
(static_cast<uint8_t>(ptr[1]) << 16) |
(static_cast<uint8_t>(ptr[2]) << 8) |
static_cast<uint8_t>(ptr[3]);
ptr += 4;
uint8_t flag = static_cast<uint8_t>(*ptr);
ptr += 1;
bool is_tombstone = (flag != 0);
if (ptr + len > end) break;
std::string key(ptr, len);
ptr += len;
key_states[key] = is_tombstone; // Last occurrence wins
}
}
bool PostingListMergeOperator::FullMergeV2(
const MergeOperationInput& merge_in,
MergeOperationOutput* merge_out) const {
// Build a map of key -> is_tombstone (last occurrence wins)
std::unordered_map<std::string, bool> key_states;
// Parse existing value (if any) - it should be a posting list binary
if (merge_in.existing_value != nullptr && !merge_in.existing_value->empty()) {
ParseExistingValue(*merge_in.existing_value, key_states);
}
// Process each operand - they are in Erlang external term format
for (const auto& operand : merge_in.operand_list) {
std::string key;
bool is_tombstone;
// Try to parse as Erlang term format first
if (ParseOperand(operand, key, is_tombstone)) {
key_states[key] = is_tombstone; // Last occurrence wins
} else {
// If parsing fails, the operand might be a raw posting list binary
// (from PartialMergeMulti output or direct storage)
// Parse it as a posting list and merge into key_states
ParseExistingValue(operand, key_states);
}
}
// Rebuild the value with only active keys (not tombstoned)
merge_out->new_value.clear();
for (const auto& [key, is_tombstone] : key_states) {
if (!is_tombstone) {
AppendEntry(&merge_out->new_value, key, false);
}
}
return true;
}
bool PostingListMergeOperator::ParseOperand(
const rocksdb::Slice& operand,
std::string& key,
bool& is_tombstone) const {
// Operand format: Erlang term {posting_add, Binary} or {posting_delete, Binary}
// Encoded using enif_term_to_binary on Erlang side
ErlNifEnv* env = enif_alloc_env();
if (!env) return false;
ERL_NIF_TERM term;
if (enif_binary_to_term(env, (unsigned char*)operand.data(),
operand.size(), &term, 0) == 0) {
enif_free_env(env);
return false;
}
int arity;
const ERL_NIF_TERM* tuple;
if (!enif_get_tuple(env, term, &arity, &tuple) || arity != 2) {
enif_free_env(env);
return false;
}
if (enif_is_identical(tuple[0], ATOM_POSTING_ADD)) {
is_tombstone = false;
} else if (enif_is_identical(tuple[0], ATOM_POSTING_DELETE)) {
is_tombstone = true;
} else {
enif_free_env(env);
return false;
}
ErlNifBinary bin;
if (!enif_inspect_binary(env, tuple[1], &bin)) {
enif_free_env(env);
return false;
}
key.assign((char*)bin.data, bin.size);
enif_free_env(env);
return true;
}
void PostingListMergeOperator::AppendEntry(
std::string* result,
const std::string& key,
bool is_tombstone) const {
// Format: <Len:32/big><Flag:8><Key:Len>
uint32_t len = static_cast<uint32_t>(key.size());
// Append length (big-endian)
result->push_back((len >> 24) & 0xFF);
result->push_back((len >> 16) & 0xFF);
result->push_back((len >> 8) & 0xFF);
result->push_back(len & 0xFF);
// Append flag byte
result->push_back(is_tombstone ? 1 : 0);
// Append key data
result->append(key);
}
bool PostingListMergeOperator::PartialMergeMulti(
const rocksdb::Slice& /*key*/,
const std::deque<rocksdb::Slice>& operand_list,
std::string* new_value,
rocksdb::Logger* /*logger*/) const {
// Combine multiple operands into a single consolidated posting list
// This enables tombstone cleanup during compaction when there's no base value
if (operand_list.size() < 2) {
return false; // Need at least 2 operands to merge
}
std::unordered_map<std::string, bool> key_states;
// Process each operand - can be Erlang term format or raw posting list
for (const auto& operand : operand_list) {
std::string key;
bool is_tombstone;
// Try to parse as Erlang term format first
if (ParseOperand(operand, key, is_tombstone)) {
key_states[key] = is_tombstone; // Last occurrence wins
} else {
// If parsing fails, the operand might be a raw posting list binary
// (from a previous PartialMergeMulti output)
ParseExistingValue(operand, key_states);
}
}
// Build the consolidated posting list with only active keys
new_value->clear();
for (const auto& [key, is_tombstone] : key_states) {
if (!is_tombstone) {
AppendEntry(new_value, key, false);
}
}
return true;
}
const char* PostingListMergeOperator::Name() const {
return "PostingListMergeOperator";
}
std::shared_ptr<PostingListMergeOperator> CreatePostingListMergeOperator() {
return std::make_shared<PostingListMergeOperator>();
}
}