Hawkin
/
arangodb
mirror of https://gitee.com/bigwinds/arangodb
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Backport fix for truncate selectivity (#8863)

This commit is contained in:
Simon 2019-04-29 17:10:03 +02:00 committed by Jan
parent dda5c3da88
commit 11aa075747
2 changed files with 130 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

100
arangod/RocksDBEngine/RocksDBCuckooIndexEstimator.h
View File

@ -527,7 +527,7 @@ class RocksDBCuckooIndexEstimator {
Result res = basics::catchVoidToResult([&]() -> void { Result res = basics::catchVoidToResult([&]() -> void {
std::vector<Key> inserts; std::vector<Key> inserts;
std::vector<Key> removals; std::vector<Key> removals;
// truncate will increase this sequence // truncate will increase this sequence
rocksdb::SequenceNumber ignoreSeq = 0; rocksdb::SequenceNumber ignoreSeq = 0;
while (true) { while (true) {
@ -535,68 +535,72 @@ class RocksDBCuckooIndexEstimator {
// find out if we have buffers to apply // find out if we have buffers to apply
{ {
WRITE_LOCKER(locker, _lock); WRITE_LOCKER(locker, _lock);
// check for a truncate marker {
auto it = _truncateBuffer.begin(); // sorted ASC // check for a truncate marker
while (it != _truncateBuffer.end() && *it <= commitSeq) { auto it = _truncateBuffer.begin(); // sorted ASC
ignoreSeq = *it; while (it != _truncateBuffer.end() && *it <= commitSeq) {
TRI_ASSERT(ignoreSeq != 0); ignoreSeq = *it;
foundTruncate = true; TRI_ASSERT(ignoreSeq != 0);
appliedSeq = std::max(appliedSeq, ignoreSeq); foundTruncate = true;
it = _truncateBuffer.erase(it); appliedSeq = std::max(appliedSeq, ignoreSeq);
it = _truncateBuffer.erase(it);
}
} }
TRI_ASSERT(ignoreSeq <= commitSeq);
// check for inserts // check for inserts
if (!_insertBuffers.empty()) { auto it = _insertBuffers.begin(); // sorted ASC
auto it = _insertBuffers.begin(); // sorted ASC while (it != _insertBuffers.end() && it->first <= commitSeq) {
if (it->first <= commitSeq) { if (it->first <= ignoreSeq) {
if (it->first >= ignoreSeq) { TRI_ASSERT(it->first <= appliedSeq);
inserts = std::move(it->second); it = _insertBuffers.erase(it);
TRI_ASSERT(!inserts.empty()); continue;
}
appliedSeq = std::max(appliedSeq, it->first);
_insertBuffers.erase(it);
} }
inserts = std::move(it->second);
TRI_ASSERT(!inserts.empty());
appliedSeq = std::max(appliedSeq, it->first);
_insertBuffers.erase(it);
break;
} }
// check for removals // check for removals
if (!_removalBuffers.empty()) { it = _removalBuffers.begin(); // sorted ASC
auto it = _removalBuffers.begin(); // sorted ASC while (it != _removalBuffers.end() && it->first <= commitSeq) {
if (it->first <= commitSeq) { if (it->first <= ignoreSeq) {
if (it->first >= ignoreSeq) { TRI_ASSERT(it->first <= appliedSeq);
removals = std::move(it->second); it = _removalBuffers.erase(it);
TRI_ASSERT(!removals.empty()); continue;
}
appliedSeq = std::max(appliedSeq, it->first);
_removalBuffers.erase(it);
} }
removals = std::move(it->second);
TRI_ASSERT(!removals.empty());
appliedSeq = std::max(appliedSeq, it->first);
_removalBuffers.erase(it);
break;
} }
} }
if (foundTruncate) { if (foundTruncate) {
clear(); // clear estimates clear(); // clear estimates
} }
// no inserts or removals left to apply, drop out of loop // no inserts or removals left to apply, drop out of loop
if (inserts.empty() && removals.empty()) { if (inserts.empty() && removals.empty()) {
break; break;
} }
if (!inserts.empty()) { // apply inserts
// apply inserts for (auto const& key : inserts) {
for (auto const& key : inserts) { insert(key);
insert(key);
}
inserts.clear();
} }
inserts.clear();
if (!removals.empty()) {
// apply removals // apply removals
for (auto const& key : removals) { for (auto const& key : removals) {
remove(key); remove(key);
}
removals.clear();
} }
removals.clear();
} // </while(true)> } // </while(true)>
}); });
return appliedSeq; return appliedSeq;
@ -925,8 +929,8 @@ class RocksDBCuckooIndexEstimator {
std::atomic<rocksdb::SequenceNumber> _committedSeq; std::atomic<rocksdb::SequenceNumber> _committedSeq;
std::atomic<bool> _needToPersist; std::atomic<bool> _needToPersist;
std::map<rocksdb::SequenceNumber, std::vector<Key>> _insertBuffers; std::multimap<rocksdb::SequenceNumber, std::vector<Key>> _insertBuffers;
std::map<rocksdb::SequenceNumber, std::vector<Key>> _removalBuffers; std::multimap<rocksdb::SequenceNumber, std::vector<Key>> _removalBuffers;
std::set<rocksdb::SequenceNumber> _truncateBuffer; std::set<rocksdb::SequenceNumber> _truncateBuffer;
HashKey _hasherKey; // Instance to compute the first hash function HashKey _hasherKey; // Instance to compute the first hash function

78
tests/RocksDBEngine/IndexEstimatorTest.cpp
View File

@ -279,5 +279,83 @@ TEST_CASE("IndexEstimator", "[rocksdb][indexestimator]") {
REQUIRE(0.1 == est.computeEstimate()); REQUIRE(0.1 == est.computeEstimate());
} }
SECTION("test_truncate_logic") {
rocksdb::SequenceNumber currentSeq(0);
rocksdb::SequenceNumber expected;
RocksDBCuckooIndexEstimator<uint64_t> est(2048);
RocksDBCollectionMeta meta;
// test buffering where we keep around one old blocker
for (size_t iteration = 0; iteration < 10; iteration++) {
uint64_t index = 0;
std::vector<uint64_t> toInsert(10);
std::vector<uint64_t> toRemove(0);
std::generate(toInsert.begin(), toInsert.end(),
[&index] { return ++index; });
est.bufferUpdates(++currentSeq, std::move(toInsert), std::move(toRemove));
}
// now make sure we haven't applied anything
std::string serialization;
expected = currentSeq;
auto appliedSeq = est.serialize(serialization, ++currentSeq);
serialization.clear();
REQUIRE(appliedSeq == expected);
REQUIRE(0.1 == est.computeEstimate());
// multiple turncate
est.bufferTruncate(currentSeq++);
est.bufferTruncate(currentSeq++);
est.bufferTruncate(currentSeq++);
uint64_t index = 0;
std::vector<uint64_t> toInsert(10);
std::vector<uint64_t> toRemove(0);
std::generate(toInsert.begin(), toInsert.end(),
[&index] { return ++index; });
est.bufferUpdates(++currentSeq, std::move(toInsert), std::move(toRemove));
expected = currentSeq;
// now make sure we haven't applied anything
appliedSeq = est.serialize(serialization, currentSeq);
serialization.clear();
REQUIRE(appliedSeq == expected);
REQUIRE(1.0 == est.computeEstimate());
}
SECTION("test_truncate_logic_2") {
rocksdb::SequenceNumber currentSeq(0);
RocksDBCuckooIndexEstimator<uint64_t> est(2048);
RocksDBCollectionMeta meta;
// test buffering where we keep around one old blocker
for (size_t iteration = 0; iteration < 10; iteration++) {
uint64_t index = 0;
std::vector<uint64_t> toInsert(10);
std::vector<uint64_t> toRemove(0);
std::generate(toInsert.begin(), toInsert.end(),
[&index] { return ++index; });
est.bufferUpdates(++currentSeq, std::move(toInsert), std::move(toRemove));
}
// truncate in the middle
est.bufferTruncate(++currentSeq);
auto expected = currentSeq;
std::string serialization;
auto appliedSeq = est.serialize(serialization, ++currentSeq);
serialization.clear();
REQUIRE(appliedSeq == expected);
REQUIRE(1.0 == est.computeEstimate());
appliedSeq = est.serialize(serialization, ++currentSeq);
REQUIRE(appliedSeq == expected);
REQUIRE(1.0 == est.computeEstimate());
}
// @brief generate tests // @brief generate tests
} }