2018-05-31 19:42:44 +02:00
|
|
|
// 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).
|
|
|
|
|
|
|
|
#ifndef ROCKSDB_LITE
|
|
|
|
|
|
|
|
#include "utilities/transactions/write_unprepared_txn_db.h"
|
|
|
|
#include "rocksdb/utilities/transaction_db.h"
|
2018-06-27 21:05:29 +02:00
|
|
|
#include "util/cast_util.h"
|
2018-05-31 19:42:44 +02:00
|
|
|
|
|
|
|
namespace rocksdb {
|
|
|
|
|
2018-07-07 02:17:36 +02:00
|
|
|
// Instead of reconstructing a Transaction object, and calling rollback on it,
|
|
|
|
// we can be more efficient with RollbackRecoveredTransaction by skipping
|
|
|
|
// unnecessary steps (eg. updating CommitMap, reconstructing keyset)
|
|
|
|
Status WriteUnpreparedTxnDB::RollbackRecoveredTransaction(
|
|
|
|
const DBImpl::RecoveredTransaction* rtxn) {
|
|
|
|
// TODO(lth): Reduce duplicate code with WritePrepared rollback logic.
|
|
|
|
assert(rtxn->unprepared_);
|
|
|
|
auto cf_map_shared_ptr = WritePreparedTxnDB::GetCFHandleMap();
|
|
|
|
auto cf_comp_map_shared_ptr = WritePreparedTxnDB::GetCFComparatorMap();
|
|
|
|
WriteOptions w_options;
|
|
|
|
// If we crash during recovery, we can just recalculate and rewrite the
|
|
|
|
// rollback batch.
|
|
|
|
w_options.disableWAL = true;
|
|
|
|
|
2019-01-08 23:43:24 +01:00
|
|
|
class InvalidSnapshotReadCallback : public ReadCallback {
|
|
|
|
public:
|
2019-04-12 23:36:36 +02:00
|
|
|
InvalidSnapshotReadCallback(SequenceNumber snapshot)
|
|
|
|
: ReadCallback(snapshot) {}
|
|
|
|
|
|
|
|
inline bool IsVisibleFullCheck(SequenceNumber) override {
|
|
|
|
// The seq provided as snapshot is the seq right before we have locked and
|
|
|
|
// wrote to it, so whatever is there, it is committed.
|
|
|
|
return true;
|
2019-01-08 23:43:24 +01:00
|
|
|
}
|
|
|
|
|
2019-04-12 23:36:36 +02:00
|
|
|
// Ignore the refresh request since we are confident that our snapshot seq
|
|
|
|
// is not going to be affected by concurrent compactions (not enabled yet.)
|
|
|
|
void Refresh(SequenceNumber) override {}
|
2019-01-08 23:43:24 +01:00
|
|
|
};
|
|
|
|
|
2018-07-07 02:17:36 +02:00
|
|
|
// Iterate starting with largest sequence number.
|
2019-05-15 22:14:18 +02:00
|
|
|
for (auto it = rtxn->batches_.rbegin(); it != rtxn->batches_.rend(); ++it) {
|
2018-07-07 02:17:36 +02:00
|
|
|
auto last_visible_txn = it->first - 1;
|
|
|
|
const auto& batch = it->second.batch_;
|
|
|
|
WriteBatch rollback_batch;
|
|
|
|
|
|
|
|
struct RollbackWriteBatchBuilder : public WriteBatch::Handler {
|
|
|
|
DBImpl* db_;
|
|
|
|
ReadOptions roptions;
|
2019-01-08 23:43:24 +01:00
|
|
|
InvalidSnapshotReadCallback callback;
|
2018-07-07 02:17:36 +02:00
|
|
|
WriteBatch* rollback_batch_;
|
|
|
|
std::map<uint32_t, const Comparator*>& comparators_;
|
|
|
|
std::map<uint32_t, ColumnFamilyHandle*>& handles_;
|
|
|
|
using CFKeys = std::set<Slice, SetComparator>;
|
|
|
|
std::map<uint32_t, CFKeys> keys_;
|
|
|
|
bool rollback_merge_operands_;
|
|
|
|
RollbackWriteBatchBuilder(
|
2019-04-12 23:36:36 +02:00
|
|
|
DBImpl* db, SequenceNumber snap_seq, WriteBatch* dst_batch,
|
2018-07-07 02:17:36 +02:00
|
|
|
std::map<uint32_t, const Comparator*>& comparators,
|
|
|
|
std::map<uint32_t, ColumnFamilyHandle*>& handles,
|
|
|
|
bool rollback_merge_operands)
|
|
|
|
: db_(db),
|
2019-04-12 23:36:36 +02:00
|
|
|
callback(snap_seq),
|
2019-04-02 23:43:03 +02:00
|
|
|
// disable min_uncommitted optimization
|
2018-07-07 02:17:36 +02:00
|
|
|
rollback_batch_(dst_batch),
|
|
|
|
comparators_(comparators),
|
|
|
|
handles_(handles),
|
|
|
|
rollback_merge_operands_(rollback_merge_operands) {}
|
|
|
|
|
|
|
|
Status Rollback(uint32_t cf, const Slice& key) {
|
|
|
|
Status s;
|
|
|
|
CFKeys& cf_keys = keys_[cf];
|
|
|
|
if (cf_keys.size() == 0) { // just inserted
|
|
|
|
auto cmp = comparators_[cf];
|
|
|
|
keys_[cf] = CFKeys(SetComparator(cmp));
|
|
|
|
}
|
|
|
|
auto res = cf_keys.insert(key);
|
|
|
|
if (res.second ==
|
|
|
|
false) { // second is false if a element already existed.
|
|
|
|
return s;
|
|
|
|
}
|
|
|
|
|
|
|
|
PinnableSlice pinnable_val;
|
|
|
|
bool not_used;
|
|
|
|
auto cf_handle = handles_[cf];
|
New API to get all merge operands for a Key (#5604)
Summary:
This is a new API added to db.h to allow for fetching all merge operands associated with a Key. The main motivation for this API is to support use cases where doing a full online merge is not necessary as it is performance sensitive. Example use-cases:
1. Update subset of columns and read subset of columns -
Imagine a SQL Table, a row is encoded as a K/V pair (as it is done in MyRocks). If there are many columns and users only updated one of them, we can use merge operator to reduce write amplification. While users only read one or two columns in the read query, this feature can avoid a full merging of the whole row, and save some CPU.
2. Updating very few attributes in a value which is a JSON-like document -
Updating one attribute can be done efficiently using merge operator, while reading back one attribute can be done more efficiently if we don't need to do a full merge.
----------------------------------------------------------------------------------------------------
API :
Status GetMergeOperands(
const ReadOptions& options, ColumnFamilyHandle* column_family,
const Slice& key, PinnableSlice* merge_operands,
GetMergeOperandsOptions* get_merge_operands_options,
int* number_of_operands)
Example usage :
int size = 100;
int number_of_operands = 0;
std::vector<PinnableSlice> values(size);
GetMergeOperandsOptions merge_operands_info;
db_->GetMergeOperands(ReadOptions(), db_->DefaultColumnFamily(), "k1", values.data(), merge_operands_info, &number_of_operands);
Description :
Returns all the merge operands corresponding to the key. If the number of merge operands in DB is greater than merge_operands_options.expected_max_number_of_operands no merge operands are returned and status is Incomplete. Merge operands returned are in the order of insertion.
merge_operands-> Points to an array of at-least merge_operands_options.expected_max_number_of_operands and the caller is responsible for allocating it. If the status returned is Incomplete then number_of_operands will contain the total number of merge operands found in DB for key.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/5604
Test Plan:
Added unit test and perf test in db_bench that can be run using the command:
./db_bench -benchmarks=getmergeoperands --merge_operator=sortlist
Differential Revision: D16657366
Pulled By: vjnadimpalli
fbshipit-source-id: 0faadd752351745224ee12d4ae9ef3cb529951bf
2019-08-06 23:22:34 +02:00
|
|
|
DBImpl::GetImplOptions get_impl_options;
|
|
|
|
get_impl_options.column_family = cf_handle;
|
|
|
|
get_impl_options.value = &pinnable_val;
|
|
|
|
get_impl_options.value_found = ¬_used;
|
|
|
|
get_impl_options.callback = &callback;
|
|
|
|
s = db_->GetImpl(roptions, key, get_impl_options);
|
2018-07-07 02:17:36 +02:00
|
|
|
assert(s.ok() || s.IsNotFound());
|
|
|
|
if (s.ok()) {
|
|
|
|
s = rollback_batch_->Put(cf_handle, key, pinnable_val);
|
|
|
|
assert(s.ok());
|
|
|
|
} else if (s.IsNotFound()) {
|
|
|
|
// There has been no readable value before txn. By adding a delete we
|
|
|
|
// make sure that there will be none afterwards either.
|
|
|
|
s = rollback_batch_->Delete(cf_handle, key);
|
|
|
|
assert(s.ok());
|
|
|
|
} else {
|
|
|
|
// Unexpected status. Return it to the user.
|
|
|
|
}
|
|
|
|
return s;
|
|
|
|
}
|
|
|
|
|
|
|
|
Status PutCF(uint32_t cf, const Slice& key,
|
|
|
|
const Slice& /*val*/) override {
|
|
|
|
return Rollback(cf, key);
|
|
|
|
}
|
|
|
|
|
|
|
|
Status DeleteCF(uint32_t cf, const Slice& key) override {
|
|
|
|
return Rollback(cf, key);
|
|
|
|
}
|
|
|
|
|
|
|
|
Status SingleDeleteCF(uint32_t cf, const Slice& key) override {
|
|
|
|
return Rollback(cf, key);
|
|
|
|
}
|
|
|
|
|
|
|
|
Status MergeCF(uint32_t cf, const Slice& key,
|
|
|
|
const Slice& /*val*/) override {
|
|
|
|
if (rollback_merge_operands_) {
|
|
|
|
return Rollback(cf, key);
|
|
|
|
} else {
|
|
|
|
return Status::OK();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Recovered batches do not contain 2PC markers.
|
|
|
|
Status MarkNoop(bool) override { return Status::InvalidArgument(); }
|
|
|
|
Status MarkBeginPrepare(bool) override {
|
|
|
|
return Status::InvalidArgument();
|
|
|
|
}
|
|
|
|
Status MarkEndPrepare(const Slice&) override {
|
|
|
|
return Status::InvalidArgument();
|
|
|
|
}
|
|
|
|
Status MarkCommit(const Slice&) override {
|
|
|
|
return Status::InvalidArgument();
|
|
|
|
}
|
|
|
|
Status MarkRollback(const Slice&) override {
|
|
|
|
return Status::InvalidArgument();
|
|
|
|
}
|
2019-04-12 23:36:36 +02:00
|
|
|
} rollback_handler(db_impl_, last_visible_txn, &rollback_batch,
|
2018-07-07 02:17:36 +02:00
|
|
|
*cf_comp_map_shared_ptr.get(), *cf_map_shared_ptr.get(),
|
2018-07-24 09:09:18 +02:00
|
|
|
txn_db_options_.rollback_merge_operands);
|
2018-07-07 02:17:36 +02:00
|
|
|
|
|
|
|
auto s = batch->Iterate(&rollback_handler);
|
|
|
|
if (!s.ok()) {
|
|
|
|
return s;
|
|
|
|
}
|
|
|
|
|
|
|
|
// The Rollback marker will be used as a batch separator
|
|
|
|
WriteBatchInternal::MarkRollback(&rollback_batch, rtxn->name_);
|
|
|
|
|
|
|
|
const uint64_t kNoLogRef = 0;
|
|
|
|
const bool kDisableMemtable = true;
|
|
|
|
const size_t kOneBatch = 1;
|
|
|
|
uint64_t seq_used = kMaxSequenceNumber;
|
|
|
|
s = db_impl_->WriteImpl(w_options, &rollback_batch, nullptr, nullptr,
|
|
|
|
kNoLogRef, !kDisableMemtable, &seq_used, kOneBatch);
|
|
|
|
if (!s.ok()) {
|
|
|
|
return s;
|
|
|
|
}
|
|
|
|
|
|
|
|
// If two_write_queues, we must manually release the sequence number to
|
|
|
|
// readers.
|
|
|
|
if (db_impl_->immutable_db_options().two_write_queues) {
|
|
|
|
db_impl_->SetLastPublishedSequence(seq_used);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return Status::OK();
|
|
|
|
}
|
|
|
|
|
|
|
|
Status WriteUnpreparedTxnDB::Initialize(
|
|
|
|
const std::vector<size_t>& compaction_enabled_cf_indices,
|
|
|
|
const std::vector<ColumnFamilyHandle*>& handles) {
|
|
|
|
// TODO(lth): Reduce code duplication in this function.
|
2019-08-27 19:57:28 +02:00
|
|
|
auto dbimpl = static_cast_with_check<DBImpl, DB>(GetRootDB());
|
2018-07-07 02:17:36 +02:00
|
|
|
assert(dbimpl != nullptr);
|
|
|
|
|
|
|
|
db_impl_->SetSnapshotChecker(new WritePreparedSnapshotChecker(this));
|
|
|
|
// A callback to commit a single sub-batch
|
|
|
|
class CommitSubBatchPreReleaseCallback : public PreReleaseCallback {
|
|
|
|
public:
|
|
|
|
explicit CommitSubBatchPreReleaseCallback(WritePreparedTxnDB* db)
|
|
|
|
: db_(db) {}
|
2019-04-03 00:14:41 +02:00
|
|
|
Status Callback(SequenceNumber commit_seq,
|
2019-06-10 20:47:16 +02:00
|
|
|
bool is_mem_disabled __attribute__((__unused__)), uint64_t,
|
|
|
|
size_t /*index*/, size_t /*total*/) override {
|
2018-07-07 02:17:36 +02:00
|
|
|
assert(!is_mem_disabled);
|
|
|
|
db_->AddCommitted(commit_seq, commit_seq);
|
|
|
|
return Status::OK();
|
|
|
|
}
|
|
|
|
|
|
|
|
private:
|
|
|
|
WritePreparedTxnDB* db_;
|
|
|
|
};
|
|
|
|
db_impl_->SetRecoverableStatePreReleaseCallback(
|
|
|
|
new CommitSubBatchPreReleaseCallback(this));
|
|
|
|
|
|
|
|
// PessimisticTransactionDB::Initialize
|
|
|
|
for (auto cf_ptr : handles) {
|
|
|
|
AddColumnFamily(cf_ptr);
|
|
|
|
}
|
|
|
|
// Verify cf options
|
|
|
|
for (auto handle : handles) {
|
|
|
|
ColumnFamilyDescriptor cfd;
|
|
|
|
Status s = handle->GetDescriptor(&cfd);
|
|
|
|
if (!s.ok()) {
|
|
|
|
return s;
|
|
|
|
}
|
|
|
|
s = VerifyCFOptions(cfd.options);
|
|
|
|
if (!s.ok()) {
|
|
|
|
return s;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Re-enable compaction for the column families that initially had
|
|
|
|
// compaction enabled.
|
|
|
|
std::vector<ColumnFamilyHandle*> compaction_enabled_cf_handles;
|
|
|
|
compaction_enabled_cf_handles.reserve(compaction_enabled_cf_indices.size());
|
|
|
|
for (auto index : compaction_enabled_cf_indices) {
|
|
|
|
compaction_enabled_cf_handles.push_back(handles[index]);
|
|
|
|
}
|
|
|
|
|
|
|
|
// create 'real' transactions from recovered shell transactions
|
|
|
|
auto rtxns = dbimpl->recovered_transactions();
|
2019-06-12 04:52:08 +02:00
|
|
|
std::map<SequenceNumber, SequenceNumber> ordered_seq_cnt;
|
2018-07-07 02:17:36 +02:00
|
|
|
for (auto rtxn : rtxns) {
|
|
|
|
auto recovered_trx = rtxn.second;
|
|
|
|
assert(recovered_trx);
|
|
|
|
assert(recovered_trx->batches_.size() >= 1);
|
|
|
|
assert(recovered_trx->name_.length());
|
|
|
|
|
|
|
|
// We can only rollback transactions after AdvanceMaxEvictedSeq is called,
|
|
|
|
// but AddPrepared must occur before AdvanceMaxEvictedSeq, which is why
|
|
|
|
// two iterations is required.
|
|
|
|
if (recovered_trx->unprepared_) {
|
|
|
|
continue;
|
|
|
|
}
|
|
|
|
|
|
|
|
WriteOptions w_options;
|
|
|
|
w_options.sync = true;
|
|
|
|
TransactionOptions t_options;
|
|
|
|
|
|
|
|
auto first_log_number = recovered_trx->batches_.begin()->second.log_number_;
|
2018-07-24 09:09:18 +02:00
|
|
|
auto first_seq = recovered_trx->batches_.begin()->first;
|
2018-07-07 02:17:36 +02:00
|
|
|
auto last_prepare_batch_cnt =
|
|
|
|
recovered_trx->batches_.begin()->second.batch_cnt_;
|
|
|
|
|
|
|
|
Transaction* real_trx = BeginTransaction(w_options, t_options, nullptr);
|
|
|
|
assert(real_trx);
|
|
|
|
auto wupt =
|
|
|
|
static_cast_with_check<WriteUnpreparedTxn, Transaction>(real_trx);
|
2019-07-17 00:19:45 +02:00
|
|
|
wupt->recovered_txn_ = true;
|
2018-07-07 02:17:36 +02:00
|
|
|
|
|
|
|
real_trx->SetLogNumber(first_log_number);
|
2018-07-24 09:09:18 +02:00
|
|
|
real_trx->SetId(first_seq);
|
2019-03-29 21:09:51 +01:00
|
|
|
Status s = real_trx->SetName(recovered_trx->name_);
|
2018-07-07 02:17:36 +02:00
|
|
|
if (!s.ok()) {
|
2019-07-17 00:19:45 +02:00
|
|
|
return s;
|
2018-07-07 02:17:36 +02:00
|
|
|
}
|
|
|
|
wupt->prepare_batch_cnt_ = last_prepare_batch_cnt;
|
|
|
|
|
|
|
|
for (auto batch : recovered_trx->batches_) {
|
|
|
|
const auto& seq = batch.first;
|
|
|
|
const auto& batch_info = batch.second;
|
|
|
|
auto cnt = batch_info.batch_cnt_ ? batch_info.batch_cnt_ : 1;
|
|
|
|
assert(batch_info.log_number_);
|
|
|
|
|
2019-06-12 04:52:08 +02:00
|
|
|
ordered_seq_cnt[seq] = cnt;
|
2018-07-07 02:17:36 +02:00
|
|
|
assert(wupt->unprep_seqs_.count(seq) == 0);
|
|
|
|
wupt->unprep_seqs_[seq] = cnt;
|
2019-07-17 00:19:45 +02:00
|
|
|
|
|
|
|
s = wupt->RebuildFromWriteBatch(batch_info.batch_);
|
2018-07-24 09:09:18 +02:00
|
|
|
assert(s.ok());
|
|
|
|
if (!s.ok()) {
|
2019-07-17 00:19:45 +02:00
|
|
|
return s;
|
2018-07-24 09:09:18 +02:00
|
|
|
}
|
2018-07-07 02:17:36 +02:00
|
|
|
}
|
|
|
|
|
2019-07-31 22:36:22 +02:00
|
|
|
const bool kClear = true;
|
|
|
|
wupt->InitWriteBatch(kClear);
|
2018-07-07 02:17:36 +02:00
|
|
|
|
|
|
|
real_trx->SetState(Transaction::PREPARED);
|
|
|
|
if (!s.ok()) {
|
2019-07-17 00:19:45 +02:00
|
|
|
return s;
|
2018-07-07 02:17:36 +02:00
|
|
|
}
|
|
|
|
}
|
2019-06-12 04:52:08 +02:00
|
|
|
// AddPrepared must be called in order
|
|
|
|
for (auto seq_cnt: ordered_seq_cnt) {
|
|
|
|
auto seq = seq_cnt.first;
|
|
|
|
auto cnt = seq_cnt.second;
|
|
|
|
for (size_t i = 0; i < cnt; i++) {
|
|
|
|
AddPrepared(seq + i);
|
|
|
|
}
|
|
|
|
}
|
2018-07-07 02:17:36 +02:00
|
|
|
|
|
|
|
SequenceNumber prev_max = max_evicted_seq_;
|
|
|
|
SequenceNumber last_seq = db_impl_->GetLatestSequenceNumber();
|
|
|
|
AdvanceMaxEvictedSeq(prev_max, last_seq);
|
2019-03-29 21:09:51 +01:00
|
|
|
// Create a gap between max and the next snapshot. This simplifies the logic
|
|
|
|
// in IsInSnapshot by not having to consider the special case of max ==
|
|
|
|
// snapshot after recovery. This is tested in IsInSnapshotEmptyMapTest.
|
|
|
|
if (last_seq) {
|
|
|
|
db_impl_->versions_->SetLastAllocatedSequence(last_seq + 1);
|
|
|
|
db_impl_->versions_->SetLastSequence(last_seq + 1);
|
|
|
|
db_impl_->versions_->SetLastPublishedSequence(last_seq + 1);
|
|
|
|
}
|
|
|
|
|
2019-04-12 23:36:36 +02:00
|
|
|
Status s;
|
2018-07-07 02:17:36 +02:00
|
|
|
// Rollback unprepared transactions.
|
|
|
|
for (auto rtxn : rtxns) {
|
|
|
|
auto recovered_trx = rtxn.second;
|
|
|
|
if (recovered_trx->unprepared_) {
|
|
|
|
s = RollbackRecoveredTransaction(recovered_trx);
|
|
|
|
if (!s.ok()) {
|
|
|
|
return s;
|
|
|
|
}
|
|
|
|
continue;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
if (s.ok()) {
|
|
|
|
dbimpl->DeleteAllRecoveredTransactions();
|
2019-04-12 23:36:36 +02:00
|
|
|
|
|
|
|
// Compaction should start only after max_evicted_seq_ is set AND recovered
|
|
|
|
// transactions are either added to PrepareHeap or rolled back.
|
|
|
|
s = EnableAutoCompaction(compaction_enabled_cf_handles);
|
2018-07-07 02:17:36 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
return s;
|
|
|
|
}
|
|
|
|
|
2018-05-31 19:42:44 +02:00
|
|
|
Transaction* WriteUnpreparedTxnDB::BeginTransaction(
|
2018-06-27 21:05:29 +02:00
|
|
|
const WriteOptions& write_options, const TransactionOptions& txn_options,
|
|
|
|
Transaction* old_txn) {
|
2018-05-31 19:42:44 +02:00
|
|
|
if (old_txn != nullptr) {
|
|
|
|
ReinitializeTransaction(old_txn, write_options, txn_options);
|
|
|
|
return old_txn;
|
|
|
|
} else {
|
|
|
|
return new WriteUnpreparedTxn(this, write_options, txn_options);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2018-06-27 21:05:29 +02:00
|
|
|
// Struct to hold ownership of snapshot and read callback for iterator cleanup.
|
|
|
|
struct WriteUnpreparedTxnDB::IteratorState {
|
|
|
|
IteratorState(WritePreparedTxnDB* txn_db, SequenceNumber sequence,
|
|
|
|
std::shared_ptr<ManagedSnapshot> s,
|
|
|
|
SequenceNumber min_uncommitted, WriteUnpreparedTxn* txn)
|
2019-08-05 22:30:56 +02:00
|
|
|
: callback(txn_db, sequence, min_uncommitted, txn->unprep_seqs_,
|
|
|
|
kBackedByDBSnapshot),
|
2019-07-24 19:21:18 +02:00
|
|
|
snapshot(s) {}
|
2019-04-02 23:43:03 +02:00
|
|
|
SequenceNumber MaxVisibleSeq() { return callback.max_visible_seq(); }
|
2018-06-27 21:05:29 +02:00
|
|
|
|
|
|
|
WriteUnpreparedTxnReadCallback callback;
|
|
|
|
std::shared_ptr<ManagedSnapshot> snapshot;
|
|
|
|
};
|
|
|
|
|
|
|
|
namespace {
|
|
|
|
static void CleanupWriteUnpreparedTxnDBIterator(void* arg1, void* /*arg2*/) {
|
|
|
|
delete reinterpret_cast<WriteUnpreparedTxnDB::IteratorState*>(arg1);
|
|
|
|
}
|
|
|
|
} // anonymous namespace
|
|
|
|
|
|
|
|
Iterator* WriteUnpreparedTxnDB::NewIterator(const ReadOptions& options,
|
|
|
|
ColumnFamilyHandle* column_family,
|
|
|
|
WriteUnpreparedTxn* txn) {
|
|
|
|
// TODO(lth): Refactor so that this logic is shared with WritePrepared.
|
|
|
|
constexpr bool ALLOW_BLOB = true;
|
|
|
|
constexpr bool ALLOW_REFRESH = true;
|
|
|
|
std::shared_ptr<ManagedSnapshot> own_snapshot = nullptr;
|
2019-07-23 17:04:58 +02:00
|
|
|
SequenceNumber snapshot_seq = kMaxSequenceNumber;
|
2018-06-27 21:05:29 +02:00
|
|
|
SequenceNumber min_uncommitted = 0;
|
2019-07-23 17:04:58 +02:00
|
|
|
|
|
|
|
// Currently, the Prev() iterator logic does not work well without snapshot
|
|
|
|
// validation. The logic simply iterates through values of a key in
|
|
|
|
// ascending seqno order, stopping at the first non-visible value and
|
|
|
|
// returning the last visible value.
|
|
|
|
//
|
|
|
|
// For example, if snapshot sequence is 3, and we have the following keys:
|
|
|
|
// foo: v1 1
|
|
|
|
// foo: v2 2
|
|
|
|
// foo: v3 3
|
|
|
|
// foo: v4 4
|
|
|
|
// foo: v5 5
|
|
|
|
//
|
|
|
|
// Then 1, 2, 3 will be visible, but 4 will be non-visible, so we return v3,
|
2019-07-24 19:21:18 +02:00
|
|
|
// which is the last visible value.
|
2019-07-23 17:04:58 +02:00
|
|
|
//
|
|
|
|
// For unprepared transactions, if we have snap_seq = 3, but the current
|
2019-07-24 19:21:18 +02:00
|
|
|
// transaction has unprep_seq 5, then returning the first non-visible value
|
2019-07-23 17:04:58 +02:00
|
|
|
// would be incorrect, as we should return v5, and not v3. The problem is that
|
2019-07-24 19:21:18 +02:00
|
|
|
// there are committed values at snapshot_seq < commit_seq < unprep_seq.
|
2019-07-23 17:04:58 +02:00
|
|
|
//
|
|
|
|
// Snapshot validation can prevent this problem by ensuring that no committed
|
2019-07-24 19:21:18 +02:00
|
|
|
// values exist at snapshot_seq < commit_seq, and thus any value with a
|
|
|
|
// sequence number greater than snapshot_seq must be unprepared values. For
|
|
|
|
// example, if the transaction had a snapshot at 3, then snapshot validation
|
|
|
|
// would be performed during the Put(v5) call. It would find v4, and the Put
|
|
|
|
// would fail with snapshot validation failure.
|
2019-07-23 17:04:58 +02:00
|
|
|
//
|
|
|
|
// TODO(lth): Improve Prev() logic to continue iterating until
|
2019-07-24 19:21:18 +02:00
|
|
|
// max_visible_seq, and then return the last visible value, so that this
|
2019-07-23 17:04:58 +02:00
|
|
|
// restriction can be lifted.
|
|
|
|
const Snapshot* snapshot = nullptr;
|
|
|
|
if (options.snapshot == nullptr) {
|
|
|
|
snapshot = GetSnapshot();
|
2018-06-27 21:05:29 +02:00
|
|
|
own_snapshot = std::make_shared<ManagedSnapshot>(db_impl_, snapshot);
|
2019-07-23 17:04:58 +02:00
|
|
|
} else {
|
|
|
|
snapshot = options.snapshot;
|
2018-06-27 21:05:29 +02:00
|
|
|
}
|
2019-07-23 17:04:58 +02:00
|
|
|
|
|
|
|
snapshot_seq = snapshot->GetSequenceNumber();
|
2018-06-27 21:05:29 +02:00
|
|
|
assert(snapshot_seq != kMaxSequenceNumber);
|
2019-07-23 17:04:58 +02:00
|
|
|
// Iteration is safe as long as largest_validated_seq <= snapshot_seq. We are
|
|
|
|
// guaranteed that for keys that were modified by this transaction (and thus
|
2019-07-24 19:21:18 +02:00
|
|
|
// might have unprepared values), no committed values exist at
|
2019-07-23 17:04:58 +02:00
|
|
|
// largest_validated_seq < commit_seq (or the contrapositive: any committed
|
2019-07-24 19:21:18 +02:00
|
|
|
// value must exist at commit_seq <= largest_validated_seq). This implies
|
2019-07-23 17:04:58 +02:00
|
|
|
// that commit_seq <= largest_validated_seq <= snapshot_seq or commit_seq <=
|
|
|
|
// snapshot_seq. As explained above, the problem with Prev() only happens when
|
|
|
|
// snapshot_seq < commit_seq.
|
|
|
|
//
|
|
|
|
// For keys that were not modified by this transaction, largest_validated_seq_
|
|
|
|
// is meaningless, and Prev() should just work with the existing visibility
|
|
|
|
// logic.
|
|
|
|
if (txn->largest_validated_seq_ > snapshot->GetSequenceNumber() &&
|
|
|
|
!txn->unprep_seqs_.empty()) {
|
|
|
|
ROCKS_LOG_ERROR(info_log_,
|
|
|
|
"WriteUnprepared iterator creation failed since the "
|
|
|
|
"transaction has performed unvalidated writes");
|
|
|
|
return nullptr;
|
|
|
|
}
|
|
|
|
min_uncommitted =
|
|
|
|
static_cast_with_check<const SnapshotImpl, const Snapshot>(snapshot)
|
|
|
|
->min_uncommitted_;
|
|
|
|
|
2018-06-27 21:05:29 +02:00
|
|
|
auto* cfd = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family)->cfd();
|
|
|
|
auto* state =
|
|
|
|
new IteratorState(this, snapshot_seq, own_snapshot, min_uncommitted, txn);
|
|
|
|
auto* db_iter =
|
2019-04-02 23:43:03 +02:00
|
|
|
db_impl_->NewIteratorImpl(options, cfd, state->MaxVisibleSeq(),
|
|
|
|
&state->callback, !ALLOW_BLOB, !ALLOW_REFRESH);
|
2018-06-27 21:05:29 +02:00
|
|
|
db_iter->RegisterCleanup(CleanupWriteUnpreparedTxnDBIterator, state, nullptr);
|
|
|
|
return db_iter;
|
|
|
|
}
|
|
|
|
|
2018-05-31 19:42:44 +02:00
|
|
|
} // namespace rocksdb
|
|
|
|
#endif // ROCKSDB_LITE
|