Function _parseSerializedMutation

fdbserver/RestoreLoader.actor.cpp:1086–1184

Parse the kv pair (version, serialized_mutation), which are the results parsed from log file, into (version, ) pair; Put the parsed versioned mutations into *pkvOps. Input key: [commitVersion_of_the_mutation_batch:uint64_t]; Input value: [includeVersion:uint64_t][val_length:uint32_t][encoded_list_of_mutations], where includeVersion is the serialized version in the batch commit

// [mutation1][mutation2]...[mutationk], where
//	a mutation is encoded as [type:uint32_t][keyLength:uint32_t][valueLength:uint32_t][keyContent][valueContent]
void _parseSerializedMutation(KeyRangeMap<Version>* pRangeVersions,
                              std::map<LoadingParam, VersionedMutationsMap>::iterator kvOpsIter,
                              SerializedMutationListMap* pmutationMap,
                              std::map<LoadingParam, SampledMutationsVec>::iterator samplesIter,
                              LoaderCounters* cc,
                              const RestoreAsset& asset) {
	VersionedMutationsMap& kvOps = kvOpsIter->second;
	SampledMutationsVec& samples = samplesIter->second;
	SerializedMutationListMap& mutationMap = *pmutationMap;

	TraceEvent(SevFRMutationInfo, "FastRestoreLoaderParseSerializedLogMutation")
	    .detail("BatchIndex", asset.batchIndex)
	    .detail("RestoreAsset", asset.toString());

	Arena tempArena;
	for (auto& m : mutationMap) {
		StringRef k = m.first.contents();
		StringRef val = m.second.first.contents();

		StringRefReader kReader(k, restore_corrupted_data());
		uint64_t commitVersion = kReader.consume<uint64_t>(); // Consume little Endian data
		// We have already filter the commit not in [beginVersion, endVersion) when we concatenate kv pair in log file
		ASSERT_WE_THINK(asset.isInVersionRange(commitVersion));

		StringRefReader vReader(val, restore_corrupted_data());
		vReader.consume<uint64_t>(); // Consume the includeVersion
		// TODO(xumengpanda): verify the protocol version is compatible and raise error if needed

		// Parse little endian value, confirmed it is correct!
		uint32_t val_length_decoded = vReader.consume<uint32_t>();
		ASSERT(val_length_decoded == val.size() - sizeof(uint64_t) - sizeof(uint32_t));

		int sub = 0;
		while (1) {
			// stop when reach the end of the string
			if (vReader.eof()) { //|| *reader.rptr == 0xFF
				break;
			}

			uint32_t type = vReader.consume<uint32_t>();
			uint32_t kLen = vReader.consume<uint32_t>();
			uint32_t vLen = vReader.consume<uint32_t>();
			const uint8_t* k = vReader.consume(kLen);
			const uint8_t* v = vReader.consume(vLen);

			MutationRef mutation((MutationRef::Type)type, KeyRef(k, kLen), KeyRef(v, vLen));
			// Should this mutation be skipped?
			// Skip mutation whose commitVesion < range kv's version
			if (logMutationTooOld(pRangeVersions, mutation, commitVersion)) {
				cc->oldLogMutations += 1;
				continue;
			}

			if (mutation.param1 >= asset.range.end ||
			    (isRangeMutation(mutation) && mutation.param2 < asset.range.begin) ||
			    (!isRangeMutation(mutation) && mutation.param1 < asset.range.begin)) {
				continue;
			}