wbtiger
/
foundationdb
mirror of https://github.com/apple/foundationdb.git
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
foundationdb/fdbserver/logsystem/ApplyMetadataMutation.cpp

1400 lines
52 KiB
C++
Raw Permalink Blame History

/*
* ApplyMetadataMutation.cpp
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2026 Apple Inc. and the FoundationDB project authors
*
* Licensed 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 "fdbclient/BackupAgent.h"
#include "fdbclient/KeyBackedTypes.h" // for key backed map codecs for tss mapping
#include "fdbclient/MutationList.h"
#include "fdbclient/Notified.h"
#include "fdbclient/SystemData.h"
#include "fdbserver/core/AccumulativeChecksumUtil.h"
#include "fdbserver/kvstore/IKeyValueStore.h"
#include "fdbserver/core/Knobs.h"
#include "fdbserver/core/LogProtocolMessage.h"
#include "fdbserver/logsystem/ApplyMetadataMutation.h"
#include "fdbserver/logsystem/LogSystem.h"
#include "flow/Error.h"
#include "flow/Trace.h"
#include "flow/UnitTest.h"
Reference<StorageInfo> getStorageInfo(UID id,
std::map<UID, Reference<StorageInfo>>* storageCache,
IKeyValueStore* txnStateStore) {
Reference<StorageInfo> storageInfo;
auto cacheItr = storageCache->find(id);
if (cacheItr == storageCache->end()) {
storageInfo = makeReference<StorageInfo>();
storageInfo->tag = decodeServerTagValue(txnStateStore->readValue(serverTagKeyFor(id)).get().get());
storageInfo->interf = decodeServerListValue(txnStateStore->readValue(serverListKeyFor(id)).get().get());
(*storageCache)[id] = storageInfo;
} else {
storageInfo = cacheItr->second;
}
return storageInfo;
}
CDCRoutingTable::CDCRoutingTable() {
tagsByRange.insert(allKeys, std::set<Tag>());
}
void CDCRoutingTable::updateRange(CDCStreamId streamId, KeyRangeRef const& keys) {
streams[streamId].keys = KeyRange(keys);
}
bool CDCRoutingTable::updateTag(CDCStreamId streamId, Version version, Tag tag) {
ASSERT_EQ(tag.locality, tagLocalityCDC);
auto& existing = streams[streamId].tag;
if (!existing.present() || version >= existing.get().first) {
existing = std::make_pair(version, tag);
return true;
}
return false;
}
void CDCRoutingTable::rebuildRanges() {
tagsByRange.insert(allKeys, std::set<Tag>());
for (const auto& [streamId, state] : streams) {
if (!state.keys.present() || !state.tag.present()) {
continue;
}
for (auto range : tagsByRange.modify(state.keys.get())) {
range->value().insert(state.tag.get().second);
}
}
tagsByRange.coalesce(allKeys);
}
void CDCRoutingTable::setRange(CDCStreamId streamId, KeyRangeRef const& keys) {
updateRange(streamId, keys);
rebuildRanges();
}
void CDCRoutingTable::setTag(CDCStreamId streamId, Version version, Tag tag) {
if (updateTag(streamId, version, tag)) {
rebuildRanges();
}
}
void CDCRoutingTable::reload(IKeyValueStore* txnStateStore) {
streams.clear();
const RangeResult streamRows = txnStateStore->readRange(cdcStreamKeys).get();
for (const auto& kv : streamRows) {
updateRange(decodeCDCStreamKey(kv.key), decodeCDCStreamKeysValue(kv.value));
}
const RangeResult tagHistoryRows = txnStateStore->readRange(cdcTagHistoryKeys).get();
for (const auto& kv : tagHistoryRows) {
const CDCTagHistoryEntry history = decodeCDCTagHistoryKey(kv.key);
updateTag(history.streamId, history.version, history.tag);
}
rebuildRanges();
}
const std::set<Tag>& CDCRoutingTable::tagsForKey(KeyRef const& key) const {
return tagsByRange.rangeContaining(key).value();
}
std::set<Tag> CDCRoutingTable::tagsForRange(KeyRangeRef const& keys) const {
std::set<Tag> tags;
for (auto range : tagsByRange.intersectingRanges(keys)) {
tags.insert(range.value().begin(), range.value().end());
}
return tags;
}
namespace {
// It is incredibly important that any modifications to txnStateStore are done in such a way that the same operations
// will be done on all commit proxies at the same time. Otherwise, the data stored in txnStateStore will become
// corrupted.
class ApplyMetadataMutationsImpl {
public:
ApplyMetadataMutationsImpl(const SpanContext& spanContext_,
const UID& dbgid_,
Arena& arena_,
const VectorRef<MutationRef>& mutations_,
IKeyValueStore* txnStateStore_)
: spanContext(spanContext_), dbgid(dbgid_), arena(arena_), mutations(mutations_), txnStateStore(txnStateStore_),
confChange(dummyConfChange), epoch(Optional<LogEpoch>()) {}
ApplyMetadataMutationsImpl(const SpanContext& spanContext_,
Arena& arena_,
const VectorRef<MutationRef>& mutations_,
const ApplyMetadataProxyContext& proxyMetadata_,
Reference<LogSystemConsumer> logSystemConsumer_,
LogPushData* toCommit_,
bool& confChange_,
Version version,
Version popVersion_,
bool initialCommit_,
bool provisionalCommitProxy_)
: spanContext(spanContext_), dbgid(proxyMetadata_.dbgid), arena(arena_), mutations(mutations_),
txnStateStore(proxyMetadata_.txnStateStore), toCommit(toCommit_), confChange(confChange_),
logSystemConsumer(logSystemConsumer_), version(version), popVersion(popVersion_),
vecBackupKeys(proxyMetadata_.vecBackupKeys), cdcRouting(proxyMetadata_.cdcRouting),
keyInfo(proxyMetadata_.keyInfo), uid_applyMutationsData(proxyMetadata_.uid_applyMutationsData),
commit(proxyMetadata_.commit), cx(proxyMetadata_.cx), committedVersion(proxyMetadata_.committedVersion),
storageCache(proxyMetadata_.storageCache), tag_popped(proxyMetadata_.tag_popped),
tssMapping(proxyMetadata_.tssMapping), initialCommit(initialCommit_),
provisionalCommitProxy(provisionalCommitProxy_),
accumulativeChecksumIndex(getCommitProxyAccumulativeChecksumIndex(proxyMetadata_.commitProxyIndex)),
acsBuilder(proxyMetadata_.acsBuilder), epoch(proxyMetadata_.epoch), rangeLock(proxyMetadata_.rangeLock) {
// If commit proxy, epoch must be set
ASSERT(toCommit == nullptr || epoch.present());
}
ApplyMetadataMutationsImpl(const SpanContext& spanContext_,
ResolverData& resolverData_,
const VectorRef<MutationRef>& mutations_)
: spanContext(spanContext_), dbgid(resolverData_.dbgid), arena(resolverData_.arena), mutations(mutations_),
txnStateStore(resolverData_.txnStateStore), toCommit(resolverData_.toCommit),
confChange(resolverData_.confChanges), logSystemConsumer(resolverData_.logSystemConsumer),
popVersion(resolverData_.popVersion), keyInfo(resolverData_.keyInfo), storageCache(resolverData_.storageCache),
initialCommit(resolverData_.initialCommit), forResolver(true),
accumulativeChecksumIndex(resolverAccumulativeChecksumIndex), epoch(Optional<LogEpoch>()) {}
private:
// The following variables are incoming parameters
const SpanContext& spanContext;
const UID& dbgid;
Arena& arena;
const VectorRef<MutationRef>& mutations;
// Transaction KV store
IKeyValueStore* txnStateStore;
// non-null if these mutations were part of a new commit handled by this commit proxy
LogPushData* toCommit = nullptr;
// Flag indicates if the configure is changed
bool& confChange;
Reference<LogSystemConsumer> logSystemConsumer = Reference<LogSystemConsumer>();
Version version = invalidVersion;
Version popVersion = 0;
KeyRangeMap<std::set<Key>>* vecBackupKeys = nullptr;
CDCRoutingTable* cdcRouting = nullptr;
KeyRangeMap<ServerCacheInfo>* keyInfo = nullptr;
std::map<Key, ApplyMutationsData>* uid_applyMutationsData = nullptr;
PublicRequestStream<CommitTransactionRequest> commit = PublicRequestStream<CommitTransactionRequest>();
Database cx = Database();
NotifiedVersion* committedVersion = nullptr;
std::map<UID, Reference<StorageInfo>>* storageCache = nullptr;
std::map<Tag, Version>* tag_popped = nullptr;
std::unordered_map<UID, StorageServerInterface>* tssMapping = nullptr;
// true if the mutations were already written to the txnStateStore as part of recovery
bool initialCommit = false;
// true if called from Resolver
bool forResolver = false;
// true if called from a provisional commit proxy
bool provisionalCommitProxy = false;
// indicate which commit proxy / resolver applies mutations
uint16_t accumulativeChecksumIndex = invalidAccumulativeChecksumIndex;
std::shared_ptr<AccumulativeChecksumBuilder> acsBuilder = nullptr;
Optional<LogEpoch> epoch;
private:
// The following variables are used internally
// Testing Storage Server removal (clearing serverTagKey) needs to read tss server list value to determine it is a
// tss + find partner's tag to send the private mutation. Since the removeStorageServer transaction clears both the
// storage list and server tag, we have to enforce ordering, processing the server tag first, and postpone the
// server list clear until the end;
std::vector<KeyRangeRef> tssServerListToRemove;
// Similar to tssServerListToRemove, the TSS mapping change key needs to read the server list at the end of the
// commit
std::vector<std::pair<UID, UID>> tssMappingToAdd;
ApplyMetadataRangeLock* rangeLock = nullptr;
private:
bool dummyConfChange = false;
private:
void writeMutation(const MutationRef& m) { toCommit->writeTypedMessage(m); }
void checkSetRangeLockPrefix(const MutationRef& m) {
if (!m.param1.startsWith(rangeLockPrefix)) {
return;
} else if (rangeLock == nullptr) {
TraceEvent(SevWarnAlways, "MutationHasRangeLockPrefixButFeatureIsOff")
.detail("Mutation", m.toString())
.detail("FeatureFlag", SERVER_KNOBS->ENABLE_READ_LOCK_ON_RANGE);
return;
}
ASSERT(!initialCommit);
// RangeLock is upated by KrmSetRange which updates a range with two successive mutations
if (rangeLock->pendingRequest()) {
// The second mutation
Key endKey = m.param1.removePrefix(rangeLockPrefix);
rangeLock->consumePendingRequest(endKey);
} else {
// The first mutation
RangeLockStateSet lockSetState = m.param2.empty() ? RangeLockStateSet() : decodeRangeLockStateSet(m.param2);
Key startKey = m.param1.removePrefix(rangeLockPrefix);
rangeLock->setPendingRequest(startKey, lockSetState);
}
txnStateStore->set(KeyValueRef(m.param1, m.param2));
return;
}
void checkSetKeyServersPrefix(MutationRef m) {
if (!m.param1.startsWith(keyServersPrefix)) {
return;
}
if (!initialCommit)
txnStateStore->set(KeyValueRef(m.param1, m.param2));
if (!keyInfo) {
return;
}
KeyRef k = m.param1.removePrefix(keyServersPrefix);
if (k == allKeys.end) {
return;
}
KeyRef end = keyInfo->rangeContaining(k).end();
KeyRangeRef insertRange(k, end);
std::vector<UID> src, dest;
// txnStateStore is always an in-memory KVS, and must always be recovered before
// applyMetadataMutations is called, so a wait here should never be needed.
Future<RangeResult> fResult = txnStateStore->readRange(serverTagKeys);
decodeKeyServersValue(fResult.get(), m.param2, src, dest);
ASSERT(storageCache);
ServerCacheInfo info;
info.tags.reserve(src.size() + dest.size());
info.src_info.reserve(src.size());
info.dest_info.reserve(dest.size());
for (const auto& id : src) {
auto storageInfo = getStorageInfo(id, storageCache, txnStateStore);
ASSERT(!storageInfo->interf.isTss());
ASSERT(storageInfo->tag != invalidTag);
info.tags.push_back(storageInfo->tag);
info.src_info.push_back(storageInfo);
}
for (const auto& id : dest) {
auto storageInfo = getStorageInfo(id, storageCache, txnStateStore);
ASSERT(!storageInfo->interf.isTss());
ASSERT(storageInfo->tag != invalidTag);
info.tags.push_back(storageInfo->tag);
info.dest_info.push_back(storageInfo);
}
uniquify(info.tags);
keyInfo->insert(insertRange, info);
if (toCommit && SERVER_KNOBS->ENABLE_VERSION_VECTOR_TLOG_UNICAST) {
toCommit->setLogsChanged();
}
}
void checkSetServerKeysPrefix(MutationRef m) {
if (!m.param1.startsWith(serverKeysPrefix)) {
return;
}
if (toCommit) {
Tag tag = decodeServerTagValue(
txnStateStore->readValue(serverTagKeyFor(serverKeysDecodeServer(m.param1))).get().get());
MutationRef privatized = m;
privatized.clearChecksumAndAccumulativeIndex();
privatized.param1 = m.param1.withPrefix(systemKeys.begin, arena);
TraceEvent(SevDebug, "SendingPrivateMutation", dbgid)
.detail("Original", m)
.detail("Privatized", privatized)
.detail("Server", serverKeysDecodeServer(m.param1))
.detail("TagKey", serverTagKeyFor(serverKeysDecodeServer(m.param1)))
.detail("Tag", tag.toString());
if (acsBuilder != nullptr) {
updateMutationWithAcsAndAddMutationToAcsBuilder(
acsBuilder, privatized, tag, accumulativeChecksumIndex, epoch.get(), version, dbgid);
}
toCommit->addTag(tag);
writeMutation(privatized);
}
}
void checkSetServerTagsPrefix(MutationRef m) {
if (!m.param1.startsWith(serverTagPrefix)) {
return;
}
UID id = decodeServerTagKey(m.param1);
Tag tag = decodeServerTagValue(m.param2);
// At this point, this tag will be visible to others
// So, acsBuilder should create an brand new acsState for this tag
// If there exists an old acsState, overwite it
if (acsBuilder != nullptr) {
acsBuilder->newTag(tag, id, version);
}
if (toCommit) {
MutationRef privatized = m;
privatized.clearChecksumAndAccumulativeIndex();
privatized.param1 = m.param1.withPrefix(systemKeys.begin, arena);
TraceEvent("ServerTag", dbgid).detail("Server", id).detail("Tag", tag.toString());
TraceEvent(SevDebug, "SendingPrivatized_ServerTag", dbgid).detail("M", "LogProtocolMessage");
if (acsBuilder != nullptr) {
updateMutationWithAcsAndAddMutationToAcsBuilder(
acsBuilder, privatized, tag, accumulativeChecksumIndex, epoch.get(), version, dbgid);
}
toCommit->addTag(tag);
toCommit->writeTypedMessage(LogProtocolMessage(), true);
TraceEvent(SevDebug, "SendingPrivatized_ServerTag", dbgid).detail("M", privatized);
toCommit->addTag(tag);
writeMutation(privatized);
}
if (!initialCommit) {
txnStateStore->set(KeyValueRef(m.param1, m.param2));
if (storageCache) {
auto cacheItr = storageCache->find(id);
if (cacheItr == storageCache->end()) {
auto storageInfo = makeReference<StorageInfo>();
storageInfo->tag = tag;
Optional<Key> interfKey = txnStateStore->readValue(serverListKeyFor(id)).get();
if (interfKey.present()) {
storageInfo->interf = decodeServerListValue(interfKey.get());
}
(*storageCache)[id] = storageInfo;
} else {
cacheItr->second->tag = tag;
// These tag vectors will be repopulated by the proxy when it detects their sizes are 0.
for (auto& it : keyInfo->ranges()) {
it.value().tags.clear();
}
}
}
}
}
void checkSetConfigKeys(MutationRef m) {
if (!m.param1.startsWith(configKeysPrefix) && m.param1 != coordinatorsKey &&
m.param1 != previousCoordinatorsKey) {
return;
}
if (Optional<StringRef>(m.param2) !=
txnStateStore->readValue(m.param1)
.get()
.castTo<StringRef>()) { // FIXME: Make this check more specific, here or by reading
// configuration whenever there is a change
if ((!m.param1.startsWith(excludedServersPrefix) && m.param1 != excludedServersVersionKey) &&
(!m.param1.startsWith(failedServersPrefix) && m.param1 != failedServersVersionKey) &&
(!m.param1.startsWith(excludedLocalityPrefix) && m.param1 != excludedLocalityVersionKey) &&
(!m.param1.startsWith(failedLocalityPrefix) && m.param1 != failedLocalityVersionKey)) {
auto t = txnStateStore->readValue(m.param1).get();
TraceEvent("MutationRequiresRestart", dbgid)
.detail("M", m)
.detail("PrevValue", t.orDefault("(none)"_sr))
.detail("ToCommit", toCommit != nullptr)
.detail("InitialCommit", initialCommit);
confChange = true;
}
}
if (!initialCommit)
txnStateStore->set(KeyValueRef(m.param1, m.param2));
}
void checkSetServerListPrefix(MutationRef m) {
if (!m.param1.startsWith(serverListPrefix)) {
return;
}
if (!initialCommit) {
txnStateStore->set(KeyValueRef(m.param1, m.param2));
if (storageCache) {
UID id = decodeServerListKey(m.param1);
StorageServerInterface interf = decodeServerListValue(m.param2);
auto cacheItr = storageCache->find(id);
if (cacheItr == storageCache->end()) {
auto storageInfo = makeReference<StorageInfo>();
storageInfo->interf = interf;
Optional<Key> tagKey = txnStateStore->readValue(serverTagKeyFor(id)).get();
if (tagKey.present()) {
storageInfo->tag = decodeServerTagValue(tagKey.get());
}
(*storageCache)[id] = storageInfo;
} else {
cacheItr->second->interf = interf;
}
}
}
}
void checkSetTSSMappingKeys(MutationRef m) {
if (!m.param1.startsWith(tssMappingKeys.begin)) {
return;
}
// Normally uses key backed map, so have to use same unpacking code here.
UID ssId = TupleCodec<UID>::unpack(m.param1.removePrefix(tssMappingKeys.begin));
UID tssId = TupleCodec<UID>::unpack(m.param2);
if (!initialCommit) {
txnStateStore->set(KeyValueRef(m.param1, m.param2));
}
if (tssMapping) {
tssMappingToAdd.push_back(std::pair(ssId, tssId));
}
if (toCommit) {
// send private mutation to SS that it now has a TSS pair
MutationRef privatized = m;
privatized.clearChecksumAndAccumulativeIndex();
privatized.param1 = m.param1.withPrefix(systemKeys.begin, arena);
Optional<Value> tagV = txnStateStore->readValue(serverTagKeyFor(ssId)).get();
if (tagV.present()) {
TraceEvent(SevDebug, "SendingPrivatized_TSSID", dbgid).detail("M", privatized);
if (acsBuilder != nullptr) {
updateMutationWithAcsAndAddMutationToAcsBuilder(acsBuilder,
privatized,
decodeServerTagValue(tagV.get()),
accumulativeChecksumIndex,
epoch.get(),
version,
dbgid);
}
toCommit->addTag(decodeServerTagValue(tagV.get()));
writeMutation(privatized);
}
}
}
void checkSetTSSQuarantineKeys(MutationRef m) {
if (!m.param1.startsWith(tssQuarantineKeys.begin) || initialCommit) {
return;
}
txnStateStore->set(KeyValueRef(m.param1, m.param2));
if (!toCommit) {
return;
}
UID tssId = decodeTssQuarantineKey(m.param1);
Optional<Value> ssiV = txnStateStore->readValue(serverListKeyFor(tssId)).get();
if (!ssiV.present()) {
return;
}
StorageServerInterface ssi = decodeServerListValue(ssiV.get());
if (!ssi.isTss()) {
return;
}
Optional<Value> tagV = txnStateStore->readValue(serverTagKeyFor(ssi.tssPairID.get())).get();
if (tagV.present()) {
MutationRef privatized = m;
privatized.clearChecksumAndAccumulativeIndex();
privatized.param1 = m.param1.withPrefix(systemKeys.begin, arena);
TraceEvent(SevDebug, "SendingPrivatized_TSSQuarantine", dbgid).detail("M", privatized);
if (acsBuilder != nullptr) {
updateMutationWithAcsAndAddMutationToAcsBuilder(acsBuilder,
privatized,
decodeServerTagValue(tagV.get()),
accumulativeChecksumIndex,
epoch.get(),
version,
dbgid);
}
toCommit->addTag(decodeServerTagValue(tagV.get()));
writeMutation(privatized);
}
}
void checkSetApplyMutationsEndRange(MutationRef m) {
// only proceed when see mutation with applyMutationsEndRange
if (!m.param1.startsWith(applyMutationsEndRange.begin)) {
return;
}
if (!initialCommit)
txnStateStore->set(KeyValueRef(m.param1, m.param2));
if (uid_applyMutationsData == nullptr) {
return;
}
Key uid = m.param1.removePrefix(applyMutationsEndRange.begin);
auto& p = (*uid_applyMutationsData)[uid];
p.endVersion = BinaryReader::fromStringRef<Version>(m.param2, Unversioned());
if (p.keyVersion == Reference<KeyRangeMap<Version>>())
p.keyVersion = makeReference<KeyRangeMap<Version>>();
if (p.worker.isValid() && !p.worker.isReady()) {
return;
}
auto addPrefixValue = txnStateStore->readValue(uid.withPrefix(applyMutationsAddPrefixRange.begin)).get();
auto removePrefixValue = txnStateStore->readValue(uid.withPrefix(applyMutationsRemovePrefixRange.begin)).get();
auto beginValue = txnStateStore->readValue(uid.withPrefix(applyMutationsBeginRange.begin)).get();
// TraceEvent("BackupAgentBaseApplyMutationsBegin")
// .detail("BeginVersion",
// beginValue.present() ? BinaryReader::fromStringRef<Version>(beginValue.get(), Unversioned()) : 0)
// .detail("EndVersion", p.endVersion)
// .log();
p.worker = applyMutations(
cx,
uid,
addPrefixValue.present() ? addPrefixValue.get() : Key(),
removePrefixValue.present() ? removePrefixValue.get() : Key(),
beginValue.present() ? BinaryReader::fromStringRef<Version>(beginValue.get(), Unversioned()) : 0,
&p.endVersion,
commit,
committedVersion,
p.keyVersion,
provisionalCommitProxy);
}
void checkSetApplyMutationsKeyVersionMapRange(MutationRef m) {
if (!m.param1.startsWith(applyMutationsKeyVersionMapRange.begin)) {
return;
}
if (!initialCommit)
txnStateStore->set(KeyValueRef(m.param1, m.param2));
if (uid_applyMutationsData == nullptr) {
return;
}
if (m.param1.size() >= applyMutationsKeyVersionMapRange.begin.size() + sizeof(UID)) {
Key uid = m.param1.substr(applyMutationsKeyVersionMapRange.begin.size(), sizeof(UID));
Key k = m.param1.substr(applyMutationsKeyVersionMapRange.begin.size() + sizeof(UID));
auto& p = (*uid_applyMutationsData)[uid];
if (p.keyVersion == Reference<KeyRangeMap<Version>>())
p.keyVersion = makeReference<KeyRangeMap<Version>>();
p.keyVersion->rawInsert(k, BinaryReader::fromStringRef<Version>(m.param2, Unversioned()));
}
}
void checkSetLogRangesRange(MutationRef m) {
if (!m.param1.startsWith(logRangesRange.begin)) {
return;
}
if (!initialCommit)
txnStateStore->set(KeyValueRef(m.param1, m.param2));
if (toCommit && SERVER_KNOBS->ENABLE_VERSION_VECTOR_TLOG_UNICAST) {
toCommit->setLogsChanged();
}
if (!vecBackupKeys) {
return;
}
Key logDestination;
KeyRef logRangeBegin = logRangesDecodeKey(m.param1, nullptr);
Key logRangeEnd = logRangesDecodeValue(m.param2, &logDestination);
// Insert the logDestination into each range of vecBackupKeys overlapping the decoded range
for (auto& logRange : vecBackupKeys->modify(KeyRangeRef(logRangeBegin, logRangeEnd))) {
logRange->value().insert(logDestination);
}
for (auto& logRange : vecBackupKeys->modify(singleKeyRange(metadataVersionKey))) {
logRange->value().insert(logDestination);
}
TraceEvent("LogRangeAdd")
.detail("LogRanges", vecBackupKeys->size())
.detail("MutationKey", m.param1)
.detail("LogRangeBegin", logRangeBegin)
.detail("LogRangeEnd", logRangeEnd);
}
void checkSetCDCMetadata(MutationRef m) {
if (!cdcStreamNameKeys.contains(m.param1) && !cdcStreamKeys.contains(m.param1) &&
!cdcTagHistoryKeys.contains(m.param1) && !cdcRetiredTagPopKeys.contains(m.param1) &&
!cdcProxyKeys.contains(m.param1) && m.param1 != cdcMaxStreamIdKey &&
m.param1 != cdcProxyAssignmentChangeKey) {
return;
}
if (!initialCommit) {
txnStateStore->set(KeyValueRef(m.param1, m.param2));
}
if (toCommit && SERVER_KNOBS->ENABLE_VERSION_VECTOR_TLOG_UNICAST &&
(cdcStreamKeys.contains(m.param1) || cdcTagHistoryKeys.contains(m.param1))) {
toCommit->setLogsChanged();
}
if (!cdcRouting) {
return;
}
if (cdcStreamKeys.contains(m.param1)) {
cdcRouting->setRange(decodeCDCStreamKey(m.param1), decodeCDCStreamKeysValue(m.param2));
} else if (cdcTagHistoryKeys.contains(m.param1)) {
const CDCTagHistoryEntry history = decodeCDCTagHistoryKey(m.param1);
cdcRouting->setTag(history.streamId, history.version, history.tag);
}
}
void checkSetGlobalKeys(MutationRef m) {
if (!m.param1.startsWith(globalKeysPrefix)) {
return;
}
if (!toCommit) {
return;
}
// Notifies all servers that a Master's server epoch ends
auto allServers = txnStateStore->readRange(serverTagKeys).get();
std::set<Tag> allTags;
if (m.param1 == killStorageKey) {
int8_t safeLocality = BinaryReader::fromStringRef<int8_t>(m.param2, Unversioned());
for (auto& kv : allServers) {
Tag t = decodeServerTagValue(kv.value);
if (t.locality != safeLocality) {
allTags.insert(t);
}
}
} else {
for (auto& kv : allServers) {
allTags.insert(decodeServerTagValue(kv.value));
}
}
if (m.param1 == lastEpochEndKey) {
toCommit->addTags(allTags);
toCommit->writeTypedMessage(LogProtocolMessage(), true);
TraceEvent(SevDebug, "SendingPrivatized_GlobalKeys", dbgid).detail("M", "LogProtocolMessage");
}
MutationRef privatized = m;
privatized.clearChecksumAndAccumulativeIndex();
privatized.param1 = m.param1.withPrefix(systemKeys.begin, arena);
TraceEvent(SevDebug, "SendingPrivatized_GlobalKeys", dbgid).detail("M", privatized);
if (acsBuilder != nullptr) {
updateMutationWithAcsAndAddMutationToAcsBuilder(
acsBuilder, privatized, allTags, accumulativeChecksumIndex, epoch.get(), version, dbgid);
}
toCommit->addTags(allTags);
writeMutation(privatized);
}
// Generates private mutations for the target storage server, instructing it to create a checkpoint.
void checkSetCheckpointKeys(MutationRef m) {
if (!m.param1.startsWith(checkpointPrefix)) {
return;
}
if (toCommit) {
CheckpointMetaData checkpoint = decodeCheckpointValue(m.param2);
for (const auto& ssID : checkpoint.src) {
Optional<Value> tagValue = txnStateStore->readValue(serverTagKeyFor(ssID)).get();
if (!tagValue.present()) {
TraceEvent(SevWarn, "CheckpointServerTagNotFound", dbgid)
.detail("StorageServerID", ssID)
.detail("Checkpoint", checkpoint.toString());
continue;
}
const Tag tag = decodeServerTagValue(tagValue.get());
MutationRef privatized = m;
privatized.clearChecksumAndAccumulativeIndex();
privatized.param1 = m.param1.withPrefix(systemKeys.begin, arena);
TraceEvent("SendingPrivateMutationCheckpoint", dbgid)
.detail("Original", m)
.detail("Privatized", privatized)
.detail("Server", ssID)
.detail("TagKey", serverTagKeyFor(ssID))
.detail("Tag", tag.toString())
.detail("Checkpoint", checkpoint.toString());
if (acsBuilder != nullptr) {
updateMutationWithAcsAndAddMutationToAcsBuilder(
acsBuilder, privatized, tag, accumulativeChecksumIndex, epoch.get(), version, dbgid);
}
toCommit->addTag(tag);
writeMutation(privatized);
}
}
}
void checkSetOtherKeys(MutationRef m) {
if (initialCommit)
return;
if (m.param1 == databaseLockedKey || m.param1 == metadataVersionKey ||
m.param1 == mustContainSystemMutationsKey || m.param1.startsWith(applyMutationsBeginRange.begin) ||
m.param1.startsWith(applyMutationsAddPrefixRange.begin) ||
m.param1.startsWith(applyMutationsRemovePrefixRange.begin) || m.param1.startsWith(tagLocalityListPrefix) ||
m.param1.startsWith(serverTagHistoryPrefix) ||
m.param1.startsWith(testOnlyTxnStateStorePrefixRange.begin)) {
txnStateStore->set(KeyValueRef(m.param1, m.param2));
}
}
void checkSetMinRequiredCommitVersionKey(MutationRef m) {
if (m.param1 != minRequiredCommitVersionKey) {
return;
}
Version requested = BinaryReader::fromStringRef<Version>(m.param2, Unversioned());
TraceEvent("MinRequiredCommitVersion", dbgid).detail("Min", requested).detail("Current", popVersion);
if (!initialCommit)
txnStateStore->set(KeyValueRef(m.param1, m.param2));
confChange = true;
CODE_PROBE(true, "Recovering at a higher version.");
}
void checkSetVersionEpochKey(MutationRef m) {
if (m.param1 != versionEpochKey) {
return;
}
int64_t versionEpoch = BinaryReader::fromStringRef<int64_t>(m.param2, Unversioned());
TraceEvent("VersionEpoch", dbgid).detail("Epoch", versionEpoch);
if (!initialCommit)
txnStateStore->set(KeyValueRef(m.param1, m.param2));
confChange = true;
CODE_PROBE(true, "Setting version epoch", probe::decoration::rare);
}
void checkSetWriteRecoverKey(MutationRef m) {
if (m.param1 != writeRecoveryKey) {
return;
}
TraceEvent("WriteRecoveryKeySet", dbgid).log();
if (!initialCommit)
txnStateStore->set(KeyValueRef(m.param1, m.param2));
CODE_PROBE(true, "Snapshot created, setting writeRecoveryKey in txnStateStore", probe::decoration::rare);
}
void checkClearRangeLockPrefix(KeyRangeRef range) {
if (rangeLock == nullptr) {
return;
} else if (!rangeLockKeys.intersects(range)) {
return;
}
ASSERT(!initialCommit);
txnStateStore->clear(range & rangeLockKeys);
return;
}
void checkClearKeyServerKeys(KeyRangeRef range) {
if (!keyServersKeys.intersects(range)) {
return;
}
KeyRangeRef r = range & keyServersKeys;
if (keyInfo) {
KeyRangeRef clearRange(r.begin.removePrefix(keyServersPrefix), r.end.removePrefix(keyServersPrefix));
keyInfo->insert(clearRange,
clearRange.begin.empty() ? ServerCacheInfo()
: keyInfo->rangeContainingKeyBefore(clearRange.begin).value());
if (toCommit && SERVER_KNOBS->ENABLE_VERSION_VECTOR_TLOG_UNICAST) {
toCommit->setLogsChanged();
}
}
if (!initialCommit)
txnStateStore->clear(r);
}
void checkClearConfigKeys(MutationRef m, KeyRangeRef range) {
if (!configKeys.intersects(range)) {
return;
}
if (!initialCommit)
txnStateStore->clear(range & configKeys);
if (!excludedServersKeys.contains(range) && !failedServersKeys.contains(range) &&
!excludedLocalityKeys.contains(range) && !failedLocalityKeys.contains(range)) {
TraceEvent("MutationRequiresRestart", dbgid).detail("M", m);
confChange = true;
}
}
void checkClearServerListKeys(KeyRangeRef range) {
if (!serverListKeys.intersects(range)) {
return;
}
if (initialCommit) {
return;
}
KeyRangeRef rangeToClear = range & serverListKeys;
if (rangeToClear.singleKeyRange()) {
UID id = decodeServerListKey(rangeToClear.begin);
Optional<Value> ssiV = txnStateStore->readValue(serverListKeyFor(id)).get();
if (ssiV.present() && decodeServerListValue(ssiV.get()).isTss()) {
tssServerListToRemove.push_back(rangeToClear);
} else {
txnStateStore->clear(rangeToClear);
}
} else {
txnStateStore->clear(rangeToClear);
}
}
void checkClearTagLocalityListKeys(KeyRangeRef range) {
if (!tagLocalityListKeys.intersects(range) || initialCommit) {
return;
}
txnStateStore->clear(range & tagLocalityListKeys);
}
void checkClearServerTagKeys(MutationRef m, KeyRangeRef range) {
if (!serverTagKeys.intersects(range)) {
return;
}
// Storage server removal always happens in a separate version from any prior writes (or any subsequent
// reuse of the tag) so we can safely destroy the tag here without any concern about intra-batch
// ordering
if (logSystemConsumer && popVersion) {
auto serverKeysCleared =
txnStateStore->readRange(range & serverTagKeys).get(); // read is expected to be immediately available
for (auto& kv : serverKeysCleared) {
Tag tag = decodeServerTagValue(kv.value);
TraceEvent("ServerTagRemove")
.detail("PopVersion", popVersion)
.detail("Tag", tag.toString())
.detail("Server", decodeServerTagKey(kv.key));
if (!forResolver) {
logSystemConsumer->pop(popVersion, tag);
(*tag_popped)[tag] = popVersion;
}
ASSERT_WE_THINK(forResolver ^ (tag_popped != nullptr));
if (toCommit) {
MutationRef privatized = m;
privatized.clearChecksumAndAccumulativeIndex();
privatized.param1 = kv.key.withPrefix(systemKeys.begin, arena);
privatized.param2 = keyAfter(privatized.param1, arena);
TraceEvent(SevDebug, "SendingPrivatized_ClearServerTag", dbgid).detail("M", privatized);
if (acsBuilder != nullptr) {
updateMutationWithAcsAndAddMutationToAcsBuilder(
acsBuilder, privatized, tag, accumulativeChecksumIndex, epoch.get(), version, dbgid);
}
toCommit->addTag(tag);
writeMutation(privatized);
}
}
// Might be a tss removal, which doesn't store a tag there.
// Chained if is a little verbose, but avoids unnecessary work
if (toCommit && !initialCommit && serverKeysCleared.empty()) {
KeyRangeRef maybeTssRange = range & serverTagKeys;
if (maybeTssRange.singleKeyRange()) {
UID id = decodeServerTagKey(maybeTssRange.begin);
Optional<Value> ssiV = txnStateStore->readValue(serverListKeyFor(id)).get();
if (ssiV.present()) {
StorageServerInterface ssi = decodeServerListValue(ssiV.get());
if (ssi.isTss()) {
Optional<Value> tagV = txnStateStore->readValue(serverTagKeyFor(ssi.tssPairID.get())).get();
if (tagV.present()) {
MutationRef privatized = m;
privatized.clearChecksumAndAccumulativeIndex();
privatized.param1 = maybeTssRange.begin.withPrefix(systemKeys.begin, arena);
privatized.param2 =
keyAfter(maybeTssRange.begin, arena).withPrefix(systemKeys.begin, arena);
TraceEvent(SevDebug, "SendingPrivatized_TSSClearServerTag", dbgid)
.detail("M", privatized);
if (acsBuilder != nullptr) {
updateMutationWithAcsAndAddMutationToAcsBuilder(acsBuilder,
privatized,
decodeServerTagValue(tagV.get()),
accumulativeChecksumIndex,
epoch.get(),
version,
dbgid);
}
toCommit->addTag(decodeServerTagValue(tagV.get()));
writeMutation(privatized);
}
}
}
}
}
}
if (!initialCommit) {
KeyRangeRef clearRange = range & serverTagKeys;
txnStateStore->clear(clearRange);
if (storageCache && clearRange.singleKeyRange()) {
storageCache->erase(decodeServerTagKey(clearRange.begin));
}
}
}
void checkClearServerTagHistoryKeys(KeyRangeRef range) {
if (!serverTagHistoryKeys.intersects(range)) {
return;
}
// Once a tag has been removed from history we should pop it, since we no longer have a record of the
// tag once it has been removed from history
if (logSystemConsumer && popVersion) {
auto serverKeysCleared = txnStateStore->readRange(range & serverTagHistoryKeys)
.get(); // read is expected to be immediately available
for (auto& kv : serverKeysCleared) {
Tag tag = decodeServerTagValue(kv.value);
TraceEvent("ServerTagHistoryRemove")
.detail("PopVersion", popVersion)
.detail("Tag", tag.toString())
.detail("Version", decodeServerTagHistoryKey(kv.key));
if (!forResolver) {
logSystemConsumer->pop(popVersion, tag);
(*tag_popped)[tag] = popVersion;
}
ASSERT_WE_THINK(forResolver ^ (tag_popped != nullptr));
}
}
if (!initialCommit)
txnStateStore->clear(range & serverTagHistoryKeys);
}
void checkClearApplyMutationsEndRange(MutationRef m, KeyRangeRef range) {
if (!range.intersects(applyMutationsEndRange)) {
return;
}
KeyRangeRef commonEndRange(range & applyMutationsEndRange);
if (!initialCommit)
txnStateStore->clear(commonEndRange);
if (uid_applyMutationsData != nullptr) {
uid_applyMutationsData->erase(
uid_applyMutationsData->lower_bound(m.param1.substr(applyMutationsEndRange.begin.size())),
m.param2 == applyMutationsEndRange.end
? uid_applyMutationsData->end()
: uid_applyMutationsData->lower_bound(m.param2.substr(applyMutationsEndRange.begin.size())));
}
}
void checkClearApplyMutationKeyVersionMapRange(MutationRef m, KeyRangeRef range) {
if (!range.intersects(applyMutationsKeyVersionMapRange)) {
return;
}
KeyRangeRef commonApplyRange(range & applyMutationsKeyVersionMapRange);
if (!initialCommit)
txnStateStore->clear(commonApplyRange);
if (uid_applyMutationsData == nullptr) {
return;
}
if (m.param1.size() >= applyMutationsKeyVersionMapRange.begin.size() + sizeof(UID) &&
m.param2.size() >= applyMutationsKeyVersionMapRange.begin.size() + sizeof(UID)) {
Key uid = m.param1.substr(applyMutationsKeyVersionMapRange.begin.size(), sizeof(UID));
Key uid2 = m.param2.substr(applyMutationsKeyVersionMapRange.begin.size(), sizeof(UID));
if (uid == uid2) {
auto& p = (*uid_applyMutationsData)[uid];
if (p.keyVersion == Reference<KeyRangeMap<Version>>())
p.keyVersion = makeReference<KeyRangeMap<Version>>();
p.keyVersion->rawErase(
KeyRangeRef(m.param1.substr(applyMutationsKeyVersionMapRange.begin.size() + sizeof(UID)),
m.param2.substr(applyMutationsKeyVersionMapRange.begin.size() + sizeof(UID))));
}
}
}
void checkClearLogRangesRange(KeyRangeRef range) {
if (!range.intersects(logRangesRange)) {
return;
}
KeyRangeRef commonLogRange(range & logRangesRange);
TraceEvent("LogRangeClear")
.detail("RangeBegin", range.begin)
.detail("RangeEnd", range.end)
.detail("IntersectBegin", commonLogRange.begin)
.detail("IntersectEnd", commonLogRange.end);
if (toCommit && SERVER_KNOBS->ENABLE_VERSION_VECTOR_TLOG_UNICAST) {
toCommit->setLogsChanged();
}
// Remove the key range from the vector, if defined
if (vecBackupKeys) {
KeyRef logKeyBegin;
Key logKeyEnd, logDestination;
// Identify the backup keys being removed
// read is expected to be immediately available
auto logRangesAffected = txnStateStore->readRange(commonLogRange).get();
TraceEvent("LogRangeClearBegin").detail("AffectedLogRanges", logRangesAffected.size());
// Add the backup name to the backup locations that do not have it
for (auto logRangeAffected : logRangesAffected) {
// Parse the backup key and name
logKeyBegin = logRangesDecodeKey(logRangeAffected.key, nullptr);
// Decode the log destination and key value
logKeyEnd = logRangesDecodeValue(logRangeAffected.value, &logDestination);
TraceEvent("LogRangeErase")
.detail("AffectedKey", logRangeAffected.key)
.detail("AffectedValue", logRangeAffected.value)
.detail("LogKeyBegin", logKeyBegin)
.detail("LogKeyEnd", logKeyEnd)
.detail("LogDestination", logDestination);
// Identify the locations to place the backup key
auto logRanges = vecBackupKeys->modify(KeyRangeRef(logKeyBegin, logKeyEnd));
// Remove the log prefix from the ranges which include it
for (auto logRange : logRanges) {
auto& logRangeMap = logRange->value();
// Remove the backup name from the range
logRangeMap.erase(logDestination);
}
bool foundKey = false;
for (auto& it : vecBackupKeys->intersectingRanges(normalKeys)) {
if (it.value().contains(logDestination)) {
foundKey = true;
break;
}
}
auto& systemBackupRanges = getSystemBackupRanges();
for (auto r = systemBackupRanges.begin(); !foundKey && r != systemBackupRanges.end(); ++r) {
for (auto& it : vecBackupKeys->intersectingRanges(*r)) {
if (it.value().contains(logDestination)) {
foundKey = true;
break;
}
}
}
if (!foundKey) {
auto logRanges = vecBackupKeys->modify(singleKeyRange(metadataVersionKey));
for (auto logRange : logRanges) {
auto& logRangeMap = logRange->value();
logRangeMap.erase(logDestination);
}
}
}
// Coalesce the entire range
vecBackupKeys->coalesce(allKeys);
}
if (!initialCommit)
txnStateStore->clear(commonLogRange);
}
void checkClearCDCMetadata(KeyRangeRef range) {
if (!cdcStreamNameKeys.intersects(range) && !cdcStreamKeys.intersects(range) &&
!cdcTagHistoryKeys.intersects(range) && !cdcRetiredTagPopKeys.intersects(range) &&
!cdcProxyKeys.intersects(range) && !range.contains(cdcMaxStreamIdKey)) {
return;
}
// CDC tags may be shared and acknowledgement minima are stored outside transaction state.
// A durable retired-tag watermark lets any CDC proxy finish pops after stream removal.
if (!initialCommit) {
for (const KeyRangeRef cdcRange :
{ cdcStreamNameKeys, cdcStreamKeys, cdcTagHistoryKeys, cdcRetiredTagPopKeys, cdcProxyKeys }) {
if (cdcRange.intersects(range)) {
txnStateStore->clear(cdcRange & range);
}
}
if (range.contains(cdcMaxStreamIdKey)) {
txnStateStore->clear(singleKeyRange(cdcMaxStreamIdKey));
}
}
if (toCommit && SERVER_KNOBS->ENABLE_VERSION_VECTOR_TLOG_UNICAST &&
(cdcStreamKeys.intersects(range) || cdcTagHistoryKeys.intersects(range))) {
toCommit->setLogsChanged();
}
if (cdcRouting && (cdcStreamKeys.intersects(range) || cdcTagHistoryKeys.intersects(range))) {
cdcRouting->reload(txnStateStore);
}
}
void checkClearTssMappingKeys(MutationRef m, KeyRangeRef range) {
if (!tssMappingKeys.intersects(range)) {
return;
}
KeyRangeRef rangeToClear = range & tssMappingKeys;
ASSERT(rangeToClear.singleKeyRange());
// Normally uses key backed map, so have to use same unpacking code here.
UID ssId = TupleCodec<UID>::unpack(m.param1.removePrefix(tssMappingKeys.begin));
if (!initialCommit) {
txnStateStore->clear(rangeToClear);
}
if (tssMapping) {
tssMapping->erase(ssId);
}
if (!toCommit) {
return;
}
// send private mutation to SS to notify that it no longer has a tss pair
if (Optional<Value> tagV = txnStateStore->readValue(serverTagKeyFor(ssId)).get(); tagV.present()) {
MutationRef privatized = m;
privatized.clearChecksumAndAccumulativeIndex();
privatized.param1 = m.param1.withPrefix(systemKeys.begin, arena);
privatized.param2 = m.param2.withPrefix(systemKeys.begin, arena);
TraceEvent(SevDebug, "SendingPrivatized_ClearTSSMapping", dbgid).detail("M", privatized);
if (acsBuilder != nullptr) {
updateMutationWithAcsAndAddMutationToAcsBuilder(acsBuilder,
privatized,
decodeServerTagValue(tagV.get()),
accumulativeChecksumIndex,
epoch.get(),
version,
dbgid);
}
toCommit->addTag(decodeServerTagValue(tagV.get()));
writeMutation(privatized);
}
}
void checkClearTssQuarantineKeys(MutationRef m, KeyRangeRef range) {
if (!tssQuarantineKeys.intersects(range) || initialCommit) {
return;
}
KeyRangeRef rangeToClear = range & tssQuarantineKeys;
ASSERT(rangeToClear.singleKeyRange());
txnStateStore->clear(rangeToClear);
if (!toCommit) {
return;
}
UID tssId = decodeTssQuarantineKey(m.param1);
if (Optional<Value> ssiV = txnStateStore->readValue(serverListKeyFor(tssId)).get(); ssiV.present()) {
if (StorageServerInterface ssi = decodeServerListValue(ssiV.get()); ssi.isTss()) {
if (Optional<Value> tagV = txnStateStore->readValue(serverTagKeyFor(ssi.tssPairID.get())).get();
tagV.present()) {
MutationRef privatized = m;
privatized.clearChecksumAndAccumulativeIndex();
privatized.param1 = m.param1.withPrefix(systemKeys.begin, arena);
privatized.param2 = m.param2.withPrefix(systemKeys.begin, arena);
TraceEvent(SevDebug, "SendingPrivatized_ClearTSSQuarantine", dbgid).detail("M", privatized);
if (acsBuilder != nullptr) {
updateMutationWithAcsAndAddMutationToAcsBuilder(acsBuilder,
privatized,
decodeServerTagValue(tagV.get()),
accumulativeChecksumIndex,
epoch.get(),
version,
dbgid);
}
toCommit->addTag(decodeServerTagValue(tagV.get()));
writeMutation(privatized);
}
}
}
}
void checkClearVersionEpochKeys(MutationRef m, KeyRangeRef range) {
if (!range.contains(versionEpochKey)) {
return;
}
if (!initialCommit)
txnStateStore->clear(singleKeyRange(versionEpochKey));
TraceEvent("MutationRequiresRestart", dbgid).detail("M", m);
confChange = true;
}
void checkClearMiscRangeKeys(KeyRangeRef range) {
if (initialCommit) {
return;
}
if (range.contains(previousCoordinatorsKey)) {
txnStateStore->clear(singleKeyRange(previousCoordinatorsKey));
}
if (range.contains(coordinatorsKey)) {
txnStateStore->clear(singleKeyRange(coordinatorsKey));
}
if (range.contains(databaseLockedKey)) {
txnStateStore->clear(singleKeyRange(databaseLockedKey));
}
if (range.contains(metadataVersionKey)) {
txnStateStore->clear(singleKeyRange(metadataVersionKey));
}
if (range.contains(mustContainSystemMutationsKey)) {
txnStateStore->clear(singleKeyRange(mustContainSystemMutationsKey));
}
if (range.contains(writeRecoveryKey)) {
txnStateStore->clear(singleKeyRange(writeRecoveryKey));
}
if (range.intersects(testOnlyTxnStateStorePrefixRange)) {
txnStateStore->clear(range & testOnlyTxnStateStorePrefixRange);
}
}
public:
void apply() {
for (auto const& m : mutations) {
if (toCommit) {
toCommit->addTransactionInfo(spanContext);
}
if (m.type == MutationRef::SetValue && isSystemKey(m.param1)) {
checkSetRangeLockPrefix(m);
checkSetKeyServersPrefix(m);
checkSetServerKeysPrefix(m);
checkSetCheckpointKeys(m);
checkSetServerTagsPrefix(m);
checkSetConfigKeys(m);
checkSetServerListPrefix(m);
checkSetTSSMappingKeys(m);
checkSetTSSQuarantineKeys(m);
checkSetApplyMutationsEndRange(m);
checkSetApplyMutationsKeyVersionMapRange(m);
checkSetLogRangesRange(m);
checkSetCDCMetadata(m);
checkSetGlobalKeys(m);
checkSetWriteRecoverKey(m);
checkSetMinRequiredCommitVersionKey(m);
checkSetVersionEpochKey(m);
checkSetOtherKeys(m);
} else if (m.type == MutationRef::ClearRange && isSystemKey(m.param2)) {
KeyRangeRef range(m.param1, m.param2);
checkClearRangeLockPrefix(range);
checkClearKeyServerKeys(range);
checkClearConfigKeys(m, range);
checkClearServerListKeys(range);
checkClearTagLocalityListKeys(range);
checkClearServerTagKeys(m, range);
checkClearServerTagHistoryKeys(range);
checkClearApplyMutationsEndRange(m, range);
checkClearApplyMutationKeyVersionMapRange(m, range);
checkClearLogRangesRange(range);
checkClearCDCMetadata(range);
checkClearTssMappingKeys(m, range);
checkClearTssQuarantineKeys(m, range);
checkClearVersionEpochKeys(m, range);
checkClearMiscRangeKeys(range);
}
}
for (KeyRangeRef& range : tssServerListToRemove) {
txnStateStore->clear(range);
}
for (auto& tssPair : tssMappingToAdd) {
// read tss server list from txn state store and add it to tss mapping
StorageServerInterface tssi =
decodeServerListValue(txnStateStore->readValue(serverListKeyFor(tssPair.second)).get().get());
(*tssMapping)[tssPair.first] = tssi;
}
}
};
} // anonymous namespace
void applyMetadataMutations(SpanContext const& spanContext,
const ApplyMetadataProxyContext& proxyMetadata,
Arena& arena,
Reference<LogSystemConsumer> logSystemConsumer,
const VectorRef<MutationRef>& mutations,
LogPushData* toCommit,
bool& confChange,
Version version,
Version popVersion,
bool initialCommit,
bool provisionalCommitProxy) {
ApplyMetadataMutationsImpl(spanContext,
arena,
mutations,
proxyMetadata,
logSystemConsumer,
toCommit,
confChange,
version,
popVersion,
initialCommit,
provisionalCommitProxy)
.apply();
}
void applyMetadataMutations(SpanContext const& spanContext,
ResolverData& resolverData,
const VectorRef<MutationRef>& mutations) {
ApplyMetadataMutationsImpl(spanContext, resolverData, mutations).apply();
}
void applyMetadataMutations(SpanContext const& spanContext,
const UID& dbgid,
Arena& arena,
const VectorRef<MutationRef>& mutations,
IKeyValueStore* txnStateStore) {
ApplyMetadataMutationsImpl(spanContext, dbgid, arena, mutations, txnStateStore).apply();
}
bool containsMetadataMutation(const VectorRef<MutationRef>& mutations) {
for (auto const& m : mutations) {
if (m.type == MutationRef::SetValue && isSystemKey(m.param1)) {
if (m.param1.startsWith(globalKeysPrefix) || (m.param1.startsWith(configKeysPrefix)) ||
(m.param1.startsWith(serverListPrefix)) || (m.param1.startsWith(serverTagPrefix)) ||
(m.param1.startsWith(tssMappingKeys.begin)) || (m.param1.startsWith(tssQuarantineKeys.begin)) ||
(m.param1.startsWith(applyMutationsEndRange.begin)) ||
(m.param1.startsWith(applyMutationsKeyVersionMapRange.begin)) ||
(m.param1.startsWith(logRangesRange.begin)) || (m.param1.startsWith(serverKeysPrefix)) ||
(m.param1.startsWith(keyServersPrefix)) || cdcStreamNameKeys.contains(m.param1) ||
cdcStreamKeys.contains(m.param1) || cdcTagHistoryKeys.contains(m.param1) ||
cdcRetiredTagPopKeys.contains(m.param1) || cdcProxyKeys.contains(m.param1) ||
m.param1 == cdcMaxStreamIdKey || m.param1 == cdcProxyAssignmentChangeKey) {
return true;
}
} else if (m.type == MutationRef::ClearRange && isSystemKey(m.param2)) {
KeyRangeRef range(m.param1, m.param2);
if ((keyServersKeys.intersects(range)) || (configKeys.intersects(range)) ||
(serverListKeys.intersects(range)) || (tagLocalityListKeys.intersects(range)) ||
(serverTagKeys.intersects(range)) || (serverTagHistoryKeys.intersects(range)) ||
(range.intersects(applyMutationsEndRange)) || (range.intersects(applyMutationsKeyVersionMapRange)) ||
(range.intersects(logRangesRange)) || (tssMappingKeys.intersects(range)) ||
(tssQuarantineKeys.intersects(range)) || (range.contains(previousCoordinatorsKey)) ||
(range.contains(coordinatorsKey)) || (range.contains(databaseLockedKey)) ||
(range.contains(metadataVersionKey)) || (range.contains(mustContainSystemMutationsKey)) ||
(range.contains(writeRecoveryKey)) || (range.intersects(testOnlyTxnStateStorePrefixRange)) ||
cdcStreamNameKeys.intersects(range) || cdcStreamKeys.intersects(range) ||
cdcTagHistoryKeys.intersects(range) || cdcRetiredTagPopKeys.intersects(range) ||
cdcProxyKeys.intersects(range) || range.contains(cdcMaxStreamIdKey)) {
return true;
}
}
}
return false;
}
TEST_CASE("/NativeCDC/RoutingTable") {
CDCRoutingTable table;
const Tag ordersTag(tagLocalityCDC, 1);
const Tag overlappingTag(tagLocalityCDC, 2);
const Tag rotatedOrdersTag(tagLocalityCDC, 3);
ASSERT(table.tagsForKey("b"_sr).empty());
ASSERT(table.tagsForRange(KeyRangeRef("b"_sr, "x"_sr)).empty());
table.setRange(1, KeyRangeRef("a"_sr, "m"_sr));
table.setTag(1, 100, ordersTag);
table.setRange(2, KeyRangeRef("g"_sr, "z"_sr));
table.setTag(2, 100, overlappingTag);
ASSERT_EQ(table.tagsForKey("b"_sr), std::set<Tag>{ ordersTag });
ASSERT_EQ(table.tagsForKey("h"_sr), (std::set<Tag>{ ordersTag, overlappingTag }));
ASSERT_EQ(table.tagsForKey("x"_sr), std::set<Tag>{ overlappingTag });
ASSERT_EQ(table.tagsForRange(KeyRangeRef("b"_sr, "x"_sr)), (std::set<Tag>{ ordersTag, overlappingTag }));
table.setTag(1, 200, rotatedOrdersTag);
ASSERT_EQ(table.tagsForKey("b"_sr), std::set<Tag>{ rotatedOrdersTag });
ASSERT_EQ(table.tagsForKey("h"_sr), (std::set<Tag>{ rotatedOrdersTag, overlappingTag }));
return Void();
}
Reference in New Issue View Git Blame Copy Permalink