Packages
rocksdb
2.6.2
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/sst_file_writer.cc
// Copyright (c) 2018-2025 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 <string>
#include <memory>
#include "rocksdb/sst_file_writer.h"
#include "rocksdb/options.h"
#include "rocksdb/env.h"
#include "atoms.h"
#include "erocksdb_db.h"
#include "sst_file_writer.h"
#include "util.h"
namespace erocksdb {
ErlNifResourceType * SstFileWriterObject::m_SstFileWriter_RESOURCE(NULL);
void
SstFileWriterObject::CreateSstFileWriterType(ErlNifEnv * env)
{
ErlNifResourceFlags flags = (ErlNifResourceFlags)(ERL_NIF_RT_CREATE | ERL_NIF_RT_TAKEOVER);
m_SstFileWriter_RESOURCE = enif_open_resource_type(env, NULL, "erocksdb_SstFileWriter",
&SstFileWriterObject::SstFileWriterResourceCleanup,
flags, NULL);
return;
}
void
SstFileWriterObject::SstFileWriterResourceCleanup(ErlNifEnv * /*env*/, void * arg)
{
SstFileWriterObject* writer_ptr = (SstFileWriterObject *)arg;
writer_ptr->~SstFileWriterObject();
writer_ptr = nullptr;
return;
}
SstFileWriterObject *
SstFileWriterObject::CreateSstFileWriterResource(std::unique_ptr<rocksdb::SstFileWriter> writer)
{
SstFileWriterObject * ret_ptr;
void * alloc_ptr;
alloc_ptr = enif_alloc_resource(m_SstFileWriter_RESOURCE, sizeof(SstFileWriterObject));
ret_ptr = new (alloc_ptr) SstFileWriterObject(std::move(writer));
return ret_ptr;
}
SstFileWriterObject *
SstFileWriterObject::RetrieveSstFileWriterResource(ErlNifEnv * Env, const ERL_NIF_TERM & term)
{
SstFileWriterObject * ret_ptr;
if (!enif_get_resource(Env, term, m_SstFileWriter_RESOURCE, (void **)&ret_ptr))
return NULL;
return ret_ptr;
}
SstFileWriterObject::SstFileWriterObject(std::unique_ptr<rocksdb::SstFileWriter> writer)
: writer_(std::move(writer)) {}
SstFileWriterObject::~SstFileWriterObject()
{
if(writer_)
{
writer_.reset();
}
return;
}
rocksdb::SstFileWriter* SstFileWriterObject::writer() {
return writer_.get();
}
// Helper to get file path from Erlang term (string list or binary)
static bool get_file_path(ErlNifEnv* env, ERL_NIF_TERM term, std::string& path)
{
// Try as binary first
ErlNifBinary bin;
if (enif_inspect_binary(env, term, &bin)) {
path.assign((const char*)bin.data, bin.size);
return true;
}
// Try as iolist (includes strings)
if (enif_inspect_iolist_as_binary(env, term, &bin)) {
path.assign((const char*)bin.data, bin.size);
return true;
}
return false;
}
// Helper to make binary from std::string
static ERL_NIF_TERM make_binary_from_string(ErlNifEnv* env, const std::string& str)
{
ERL_NIF_TERM result;
unsigned char* buf = enif_make_new_binary(env, str.size(), &result);
memcpy(buf, str.data(), str.size());
return result;
}
ERL_NIF_TERM
SstFileWriterOpen(
ErlNifEnv* env,
int /*argc*/,
const ERL_NIF_TERM argv[])
{
// argv[0]: Options list
// argv[1]: FilePath (binary/string)
rocksdb::Options options;
ERL_NIF_TERM head, tail;
tail = argv[0];
// Parse options - each option is a tuple {atom, value}
while(enif_get_list_cell(env, tail, &head, &tail)) {
// parse_db_option and parse_cf_option handle the tuple internally
ERL_NIF_TERM result = parse_db_option(env, head, options);
if (result != ATOM_OK) {
result = parse_cf_option(env, head, options);
}
// Ignore unknown options to be lenient
}
// Get file path
std::string file_path;
if (!get_file_path(env, argv[1], file_path)) {
return enif_make_badarg(env);
}
// Create EnvOptions from Options
rocksdb::EnvOptions env_options;
// Create the SstFileWriter
auto sst_writer = std::make_unique<rocksdb::SstFileWriter>(env_options, options);
// Open the file for writing
rocksdb::Status status = sst_writer->Open(file_path);
if (!status.ok()) {
return error_tuple(env, ATOM_ERROR, status);
}
// Wrap in resource
auto writer_ptr = SstFileWriterObject::CreateSstFileWriterResource(std::move(sst_writer));
ERL_NIF_TERM result = enif_make_resource(env, writer_ptr);
enif_release_resource(writer_ptr);
return enif_make_tuple2(env, ATOM_OK, result);
}
ERL_NIF_TERM
SstFileWriterPut(
ErlNifEnv* env,
int /*argc*/,
const ERL_NIF_TERM argv[])
{
// argv[0]: SstFileWriter resource
// argv[1]: Key (binary)
// argv[2]: Value (binary)
SstFileWriterObject* writer_ptr = SstFileWriterObject::RetrieveSstFileWriterResource(env, argv[0]);
if (nullptr == writer_ptr) {
return enif_make_badarg(env);
}
ErlNifBinary key, value;
if (!enif_inspect_binary(env, argv[1], &key) ||
!enif_inspect_binary(env, argv[2], &value)) {
return enif_make_badarg(env);
}
rocksdb::Slice key_slice((const char*)key.data, key.size);
rocksdb::Slice value_slice((const char*)value.data, value.size);
rocksdb::Status status = writer_ptr->writer()->Put(key_slice, value_slice);
if (!status.ok()) {
return error_tuple(env, ATOM_ERROR, status);
}
return ATOM_OK;
}
ERL_NIF_TERM
SstFileWriterPutEntity(
ErlNifEnv* env,
int /*argc*/,
const ERL_NIF_TERM argv[])
{
// argv[0]: SstFileWriter resource
// argv[1]: Key (binary)
// argv[2]: Columns (list of {Name, Value} tuples)
SstFileWriterObject* writer_ptr = SstFileWriterObject::RetrieveSstFileWriterResource(env, argv[0]);
if (nullptr == writer_ptr) {
return enif_make_badarg(env);
}
ErlNifBinary key;
if (!enif_inspect_binary(env, argv[1], &key)) {
return enif_make_badarg(env);
}
rocksdb::Slice key_slice((const char*)key.data, key.size);
// Parse columns list - we need to store the data in vectors to keep slices valid
std::vector<std::string> col_names;
std::vector<std::string> col_values;
ERL_NIF_TERM head, tail;
tail = argv[2];
while(enif_get_list_cell(env, tail, &head, &tail)) {
const ERL_NIF_TERM* column;
int arity;
if (enif_get_tuple(env, head, &arity, &column) && 2 == arity) {
ErlNifBinary col_name_bin, col_value_bin;
if (!enif_inspect_binary(env, column[0], &col_name_bin) ||
!enif_inspect_binary(env, column[1], &col_value_bin)) {
return enif_make_badarg(env);
}
col_names.emplace_back((const char*)col_name_bin.data, col_name_bin.size);
col_values.emplace_back((const char*)col_value_bin.data, col_value_bin.size);
} else {
return enif_make_badarg(env);
}
}
// Build WideColumns from stored strings
rocksdb::WideColumns columns;
for (size_t i = 0; i < col_names.size(); ++i) {
columns.push_back({
rocksdb::Slice(col_names[i]),
rocksdb::Slice(col_values[i])
});
}
rocksdb::Status status = writer_ptr->writer()->PutEntity(key_slice, columns);
if (!status.ok()) {
return error_tuple(env, ATOM_ERROR, status);
}
return ATOM_OK;
}
ERL_NIF_TERM
SstFileWriterMerge(
ErlNifEnv* env,
int /*argc*/,
const ERL_NIF_TERM argv[])
{
// argv[0]: SstFileWriter resource
// argv[1]: Key (binary)
// argv[2]: Value (binary)
SstFileWriterObject* writer_ptr = SstFileWriterObject::RetrieveSstFileWriterResource(env, argv[0]);
if (nullptr == writer_ptr) {
return enif_make_badarg(env);
}
ErlNifBinary key, value;
if (!enif_inspect_binary(env, argv[1], &key) ||
!enif_inspect_binary(env, argv[2], &value)) {
return enif_make_badarg(env);
}
rocksdb::Slice key_slice((const char*)key.data, key.size);
rocksdb::Slice value_slice((const char*)value.data, value.size);
rocksdb::Status status = writer_ptr->writer()->Merge(key_slice, value_slice);
if (!status.ok()) {
return error_tuple(env, ATOM_ERROR, status);
}
return ATOM_OK;
}
ERL_NIF_TERM
SstFileWriterDelete(
ErlNifEnv* env,
int /*argc*/,
const ERL_NIF_TERM argv[])
{
// argv[0]: SstFileWriter resource
// argv[1]: Key (binary)
SstFileWriterObject* writer_ptr = SstFileWriterObject::RetrieveSstFileWriterResource(env, argv[0]);
if (nullptr == writer_ptr) {
return enif_make_badarg(env);
}
ErlNifBinary key;
if (!enif_inspect_binary(env, argv[1], &key)) {
return enif_make_badarg(env);
}
rocksdb::Slice key_slice((const char*)key.data, key.size);
rocksdb::Status status = writer_ptr->writer()->Delete(key_slice);
if (!status.ok()) {
return error_tuple(env, ATOM_ERROR, status);
}
return ATOM_OK;
}
ERL_NIF_TERM
SstFileWriterDeleteRange(
ErlNifEnv* env,
int /*argc*/,
const ERL_NIF_TERM argv[])
{
// argv[0]: SstFileWriter resource
// argv[1]: Begin key (binary)
// argv[2]: End key (binary)
SstFileWriterObject* writer_ptr = SstFileWriterObject::RetrieveSstFileWriterResource(env, argv[0]);
if (nullptr == writer_ptr) {
return enif_make_badarg(env);
}
ErlNifBinary begin_key, end_key;
if (!enif_inspect_binary(env, argv[1], &begin_key) ||
!enif_inspect_binary(env, argv[2], &end_key)) {
return enif_make_badarg(env);
}
rocksdb::Slice begin_slice((const char*)begin_key.data, begin_key.size);
rocksdb::Slice end_slice((const char*)end_key.data, end_key.size);
rocksdb::Status status = writer_ptr->writer()->DeleteRange(begin_slice, end_slice);
if (!status.ok()) {
return error_tuple(env, ATOM_ERROR, status);
}
return ATOM_OK;
}
ERL_NIF_TERM
SstFileWriterFinish(
ErlNifEnv* env,
int argc,
const ERL_NIF_TERM argv[])
{
// argv[0]: SstFileWriter resource
// argv[1]: (optional) with_file_info atom
SstFileWriterObject* writer_ptr = SstFileWriterObject::RetrieveSstFileWriterResource(env, argv[0]);
if (nullptr == writer_ptr) {
return enif_make_badarg(env);
}
bool with_file_info = false;
if (argc > 1 && argv[1] == ATOM_WITH_FILE_INFO) {
with_file_info = true;
}
if (with_file_info) {
rocksdb::ExternalSstFileInfo file_info;
rocksdb::Status status = writer_ptr->writer()->Finish(&file_info);
if (!status.ok()) {
return error_tuple(env, ATOM_ERROR, status);
}
// Build file info map
ERL_NIF_TERM keys[9];
ERL_NIF_TERM values[9];
int idx = 0;
keys[idx] = ATOM_FILE_PATH;
values[idx] = make_binary_from_string(env, file_info.file_path);
idx++;
keys[idx] = ATOM_SMALLEST_KEY;
values[idx] = make_binary_from_string(env, file_info.smallest_key);
idx++;
keys[idx] = ATOM_LARGEST_KEY;
values[idx] = make_binary_from_string(env, file_info.largest_key);
idx++;
keys[idx] = ATOM_SMALLEST_RANGE_DEL_KEY;
values[idx] = make_binary_from_string(env, file_info.smallest_range_del_key);
idx++;
keys[idx] = ATOM_LARGEST_RANGE_DEL_KEY;
values[idx] = make_binary_from_string(env, file_info.largest_range_del_key);
idx++;
keys[idx] = ATOM_FILE_SIZE;
values[idx] = enif_make_uint64(env, file_info.file_size);
idx++;
keys[idx] = ATOM_NUM_ENTRIES;
values[idx] = enif_make_uint64(env, file_info.num_entries);
idx++;
keys[idx] = ATOM_NUM_RANGE_DEL_ENTRIES;
values[idx] = enif_make_uint64(env, file_info.num_range_del_entries);
idx++;
keys[idx] = ATOM_SEQUENCE_NUMBER;
values[idx] = enif_make_uint64(env, file_info.sequence_number);
idx++;
ERL_NIF_TERM info_map;
enif_make_map_from_arrays(env, keys, values, idx, &info_map);
return enif_make_tuple2(env, ATOM_OK, info_map);
} else {
rocksdb::Status status = writer_ptr->writer()->Finish();
if (!status.ok()) {
return error_tuple(env, ATOM_ERROR, status);
}
return ATOM_OK;
}
}
ERL_NIF_TERM
SstFileWriterFileSize(
ErlNifEnv* env,
int /*argc*/,
const ERL_NIF_TERM argv[])
{
// argv[0]: SstFileWriter resource
SstFileWriterObject* writer_ptr = SstFileWriterObject::RetrieveSstFileWriterResource(env, argv[0]);
if (nullptr == writer_ptr) {
return enif_make_badarg(env);
}
uint64_t file_size = writer_ptr->writer()->FileSize();
return enif_make_uint64(env, file_size);
}
ERL_NIF_TERM
ReleaseSstFileWriter(
ErlNifEnv* env,
int /*argc*/,
const ERL_NIF_TERM argv[])
{
SstFileWriterObject* writer_ptr = SstFileWriterObject::RetrieveSstFileWriterResource(env, argv[0]);
if (nullptr == writer_ptr) {
return ATOM_OK;
}
// Resource will be cleaned up by GC
return ATOM_OK;
}
} // namespace erocksdb