Summary: add a new transaction option `TransactionOptions::commit_bypass_memtable` that will ingest the transaction into a DB as an immutable memtables, skipping memtable writes during transaction commit. This helps to reduce the blocking time of committing a large transaction, which is mostly spent on memtable writes. The ingestion is done by creating WBWIMemTable using transaction's underlying WBWI, and ingest it as the latest immutable memtable. The feature will be experimental. Major changes are: 1. write path change to ingest the transaction, mostly in WriteImpl() and IngestWBWI() in db_impl_write.cc. 2. WBWI changes to track some per CF stats like entry count and overwritten single deletion count, and track which keys have overwritten single deletions (see 3.). Per CF stat is used to precompute the number of entries in each WBWIMemTable. 3. WBWIMemTable Iterator changes to emit overwritten single deletions. The motivation is explained in the comment above class WBWIMemTable definition. The rest of the changes in WBWIMemTable are moving the iterator definition around. Some intended follow ups: 1. support for merge operations 2. stats/logging around this option 3. tests improvement, including stress test support for the more comprehensive no_batched_op_stress. Pull Request resolved: https://github.com/facebook/rocksdb/pull/13144 Test Plan: * added new unit tests * enabled in multi_ops_txns_stress test * Benchmark: applying the change in 8222c0cafc4c6eb3a0d05807f7014b44998acb7a, I tested txn size of 10k and check perf context for write_memtable_time, write_wal_time and key_lock_wait_time(repurposed for transaction unlock time). Though the benchmark result number can be flaky, this shows memtable write time improved a lot (more than 100 times). The benchmark also shows that the remaining commit latency is from transaction unlock. ``` ./db_bench --benchmarks=fillrandom --seed=1727376962 --threads=1 --disable_auto_compactions=1 --max_write_buffer_number=100 --min_write_buffer_number_to_merge=100 --writes=100000 --batch_size=10000 --transaction_db=1 --perf_level=4 --enable_pipelined_write=false --commit_bypass_memtable=1 commit_bypass_memtable = false fillrandom : 3.982 micros/op 251119 ops/sec 0.398 seconds 100000 operations; 27.8 MB/s PERF_CONTEXT: write_memtable_time = 116950422 write_wal_time = 8535565 txn unlock time = 32979883 commit_bypass_memtable = true fillrandom : 2.627 micros/op 380559 ops/sec 0.263 seconds 100000 operations; 42.1 MB/s PERF_CONTEXT: write_memtable_time = 740784 write_wal_time = 11993119 txn unlock time = 21735685 ``` Reviewed By: jowlyzhang Differential Revision: D66307632 Pulled By: cbi42 fbshipit-source-id: 6619af58c4c537aed1f76c4a7e869fb3f5098999
200 lines
7.2 KiB
C++
200 lines
7.2 KiB
C++
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
|
|
// This source code is licensed under both the GPLv2 (found in the
|
|
// COPYING file in the root directory) and Apache 2.0 License
|
|
// (found in the LICENSE.Apache file in the root directory).
|
|
//
|
|
// This file implements the "bridge" between Java and C++ and enables
|
|
// calling c++ ROCKSDB_NAMESPACE::WriteBatch methods testing from Java side.
|
|
#include "rocksdb/write_batch.h"
|
|
|
|
#include <memory>
|
|
|
|
#include "db/memtable.h"
|
|
#include "db/write_batch_internal.h"
|
|
#include "include/org_rocksdb_WriteBatch.h"
|
|
#include "include/org_rocksdb_WriteBatchTest.h"
|
|
#include "include/org_rocksdb_WriteBatchTestInternalHelper.h"
|
|
#include "include/org_rocksdb_WriteBatch_Handler.h"
|
|
#include "options/cf_options.h"
|
|
#include "rocksdb/db.h"
|
|
#include "rocksdb/env.h"
|
|
#include "rocksdb/memtablerep.h"
|
|
#include "rocksdb/status.h"
|
|
#include "rocksdb/write_buffer_manager.h"
|
|
#include "rocksjni/portal.h"
|
|
#include "test_util/testharness.h"
|
|
#include "util/string_util.h"
|
|
|
|
/*
|
|
* Class: org_rocksdb_WriteBatchTest
|
|
* Method: getContents
|
|
* Signature: (J)[B
|
|
*/
|
|
jbyteArray Java_org_rocksdb_WriteBatchTest_getContents(JNIEnv* env,
|
|
jclass /*jclazz*/,
|
|
jlong jwb_handle) {
|
|
auto* b = reinterpret_cast<ROCKSDB_NAMESPACE::WriteBatch*>(jwb_handle);
|
|
assert(b != nullptr);
|
|
|
|
// todo: Currently the following code is directly copied from
|
|
// db/write_bench_test.cc. It could be implemented in java once
|
|
// all the necessary components can be accessed via jni api.
|
|
|
|
ROCKSDB_NAMESPACE::InternalKeyComparator cmp(
|
|
ROCKSDB_NAMESPACE::BytewiseComparator());
|
|
auto factory = std::make_shared<ROCKSDB_NAMESPACE::SkipListFactory>();
|
|
ROCKSDB_NAMESPACE::Options options;
|
|
ROCKSDB_NAMESPACE::WriteBufferManager wb(options.db_write_buffer_size);
|
|
options.memtable_factory = factory;
|
|
ROCKSDB_NAMESPACE::MemTable* mem = new ROCKSDB_NAMESPACE::MemTable(
|
|
cmp, ROCKSDB_NAMESPACE::ImmutableOptions(options),
|
|
ROCKSDB_NAMESPACE::MutableCFOptions(options), &wb,
|
|
ROCKSDB_NAMESPACE::kMaxSequenceNumber, 0 /* column_family_id */);
|
|
mem->Ref();
|
|
std::string state;
|
|
ROCKSDB_NAMESPACE::ColumnFamilyMemTablesDefault cf_mems_default(mem);
|
|
ROCKSDB_NAMESPACE::Status s =
|
|
ROCKSDB_NAMESPACE::WriteBatchInternal::InsertInto(b, &cf_mems_default,
|
|
nullptr, nullptr);
|
|
unsigned int count = 0;
|
|
ROCKSDB_NAMESPACE::Arena arena;
|
|
ROCKSDB_NAMESPACE::ScopedArenaPtr<ROCKSDB_NAMESPACE::InternalIterator> iter(
|
|
mem->NewIterator(ROCKSDB_NAMESPACE::ReadOptions(),
|
|
/*seqno_to_time_mapping=*/nullptr, &arena,
|
|
/*prefix_extractor=*/nullptr, /*for_flush=*/false));
|
|
for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
|
|
ROCKSDB_NAMESPACE::ParsedInternalKey ikey;
|
|
ikey.clear();
|
|
ROCKSDB_NAMESPACE::Status pik_status = ROCKSDB_NAMESPACE::ParseInternalKey(
|
|
iter->key(), &ikey, true /* log_err_key */);
|
|
pik_status.PermitUncheckedError();
|
|
assert(pik_status.ok());
|
|
switch (ikey.type) {
|
|
case ROCKSDB_NAMESPACE::kTypeValue:
|
|
state.append("Put(");
|
|
state.append(ikey.user_key.ToString());
|
|
state.append(", ");
|
|
state.append(iter->value().ToString());
|
|
state.append(")");
|
|
count++;
|
|
break;
|
|
case ROCKSDB_NAMESPACE::kTypeMerge:
|
|
state.append("Merge(");
|
|
state.append(ikey.user_key.ToString());
|
|
state.append(", ");
|
|
state.append(iter->value().ToString());
|
|
state.append(")");
|
|
count++;
|
|
break;
|
|
case ROCKSDB_NAMESPACE::kTypeDeletion:
|
|
state.append("Delete(");
|
|
state.append(ikey.user_key.ToString());
|
|
state.append(")");
|
|
count++;
|
|
break;
|
|
case ROCKSDB_NAMESPACE::kTypeSingleDeletion:
|
|
state.append("SingleDelete(");
|
|
state.append(ikey.user_key.ToString());
|
|
state.append(")");
|
|
count++;
|
|
break;
|
|
case ROCKSDB_NAMESPACE::kTypeRangeDeletion:
|
|
state.append("DeleteRange(");
|
|
state.append(ikey.user_key.ToString());
|
|
state.append(", ");
|
|
state.append(iter->value().ToString());
|
|
state.append(")");
|
|
count++;
|
|
break;
|
|
case ROCKSDB_NAMESPACE::kTypeLogData:
|
|
state.append("LogData(");
|
|
state.append(ikey.user_key.ToString());
|
|
state.append(")");
|
|
count++;
|
|
break;
|
|
default:
|
|
assert(false);
|
|
state.append("Err:Expected(");
|
|
state.append(std::to_string(ikey.type));
|
|
state.append(")");
|
|
count++;
|
|
break;
|
|
}
|
|
state.append("@");
|
|
state.append(std::to_string(ikey.sequence));
|
|
}
|
|
if (!s.ok()) {
|
|
state.append(s.ToString());
|
|
} else if (ROCKSDB_NAMESPACE::WriteBatchInternal::Count(b) != count) {
|
|
state.append("Err:CountMismatch(expected=");
|
|
state.append(
|
|
std::to_string(ROCKSDB_NAMESPACE::WriteBatchInternal::Count(b)));
|
|
state.append(", actual=");
|
|
state.append(std::to_string(count));
|
|
state.append(")");
|
|
}
|
|
delete mem->Unref();
|
|
|
|
jbyteArray jstate = env->NewByteArray(static_cast<jsize>(state.size()));
|
|
if (jstate == nullptr) {
|
|
// exception thrown: OutOfMemoryError
|
|
return nullptr;
|
|
}
|
|
|
|
env->SetByteArrayRegion(
|
|
jstate, 0, static_cast<jsize>(state.size()),
|
|
const_cast<jbyte*>(reinterpret_cast<const jbyte*>(state.c_str())));
|
|
if (env->ExceptionCheck()) {
|
|
// exception thrown: ArrayIndexOutOfBoundsException
|
|
env->DeleteLocalRef(jstate);
|
|
return nullptr;
|
|
}
|
|
|
|
return jstate;
|
|
}
|
|
|
|
/*
|
|
* Class: org_rocksdb_WriteBatchTestInternalHelper
|
|
* Method: setSequence
|
|
* Signature: (JJ)V
|
|
*/
|
|
void Java_org_rocksdb_WriteBatchTestInternalHelper_setSequence(
|
|
JNIEnv* /*env*/, jclass /*jclazz*/, jlong jwb_handle, jlong jsn) {
|
|
auto* wb = reinterpret_cast<ROCKSDB_NAMESPACE::WriteBatch*>(jwb_handle);
|
|
assert(wb != nullptr);
|
|
|
|
ROCKSDB_NAMESPACE::WriteBatchInternal::SetSequence(
|
|
wb, static_cast<ROCKSDB_NAMESPACE::SequenceNumber>(jsn));
|
|
}
|
|
|
|
/*
|
|
* Class: org_rocksdb_WriteBatchTestInternalHelper
|
|
* Method: sequence
|
|
* Signature: (J)J
|
|
*/
|
|
jlong Java_org_rocksdb_WriteBatchTestInternalHelper_sequence(JNIEnv* /*env*/,
|
|
jclass /*jclazz*/,
|
|
jlong jwb_handle) {
|
|
auto* wb = reinterpret_cast<ROCKSDB_NAMESPACE::WriteBatch*>(jwb_handle);
|
|
assert(wb != nullptr);
|
|
|
|
return static_cast<jlong>(
|
|
ROCKSDB_NAMESPACE::WriteBatchInternal::Sequence(wb));
|
|
}
|
|
|
|
/*
|
|
* Class: org_rocksdb_WriteBatchTestInternalHelper
|
|
* Method: append
|
|
* Signature: (JJ)V
|
|
*/
|
|
void Java_org_rocksdb_WriteBatchTestInternalHelper_append(JNIEnv* /*env*/,
|
|
jclass /*jclazz*/,
|
|
jlong jwb_handle_1,
|
|
jlong jwb_handle_2) {
|
|
auto* wb1 = reinterpret_cast<ROCKSDB_NAMESPACE::WriteBatch*>(jwb_handle_1);
|
|
assert(wb1 != nullptr);
|
|
auto* wb2 = reinterpret_cast<ROCKSDB_NAMESPACE::WriteBatch*>(jwb_handle_2);
|
|
assert(wb2 != nullptr);
|
|
|
|
ROCKSDB_NAMESPACE::WriteBatchInternal::Append(wb1, wb2);
|
|
}
|