Finish getRangeAndMap at NativeAPI layer and below

And make GetRangeAndMap simulation test cover the changes

fdbclient/FDBTypes.h

@@ -475,6 +475,7 @@ using KeyRange = Standalone<KeyRangeRef>;
 using KeyValue = Standalone<KeyValueRef>;
 using KeySelector = Standalone<struct KeySelectorRef>;
 using RangeResult = Standalone<struct RangeResultRef>;
+using RangeAndMapResult = Standalone<struct RangeAndMapResultRef>;
 
 enum { invalidVersion = -1, latestVersion = -2, MAX_VERSION = std::numeric_limits<int64_t>::max() };
 
@@ -610,6 +611,8 @@ KeyRangeWith<Val> keyRangeWith(const KeyRangeRef& range, const Val& value) {
 	return KeyRangeWith<Val>(range, value);
 }
 
+struct KeyValueAndMappedReqAndResultRef;
+
 struct GetRangeLimits {
 	enum { ROW_LIMIT_UNLIMITED = -1, BYTE_LIMIT_UNLIMITED = -1 };
 
@@ -623,6 +626,8 @@ struct GetRangeLimits {
 
 	void decrement(VectorRef<KeyValueRef> const& data);
 	void decrement(KeyValueRef const& data);
+	void decrement(VectorRef<KeyValueAndMappedReqAndResultRef> const& data);
+	void decrement(KeyValueAndMappedReqAndResultRef const& data);
 
 	// True if either the row or byte limit has been reached
 	bool isReached();
@@ -683,6 +688,116 @@ struct Traceable<RangeResultRef> : std::true_type {
 	}
 };
 
+// Similar to KeyValueRef, but result can be empty.
+struct GetValueReqAndResultRef {
+	KeyRef key;
+	Optional<ValueRef> result;
+
+	GetValueReqAndResultRef() {}
+	GetValueReqAndResultRef(Arena& a, const GetValueReqAndResultRef& copyFrom)
+	  : key(a, copyFrom.key), result(a, copyFrom.result) {}
+
+	bool operator==(const GetValueReqAndResultRef& rhs) const { return key == rhs.key && result == rhs.result; }
+	bool operator!=(const GetValueReqAndResultRef& rhs) const { return !(rhs == *this); }
+	int expectedSize() const { return key.expectedSize() + result.expectedSize(); }
+
+	template <class Ar>
+	void serialize(Ar& ar) {
+		serializer(ar, key, result);
+	}
+};
+
+struct GetRangeReqAndResultRef {
+	KeySelectorRef begin, end;
+	RangeResultRef result;
+
+	GetRangeReqAndResultRef() {}
+	//	KeyValueRef(const KeyRef& key, const ValueRef& value) : key(key), value(value) {}
+	GetRangeReqAndResultRef(Arena& a, const GetRangeReqAndResultRef& copyFrom)
+	  : begin(a, copyFrom.begin), end(a, copyFrom.end), result(a, copyFrom.result) {}
+
+	bool operator==(const GetRangeReqAndResultRef& rhs) const {
+		return begin == rhs.begin && end == rhs.end && result == rhs.result;
+	}
+	bool operator!=(const GetRangeReqAndResultRef& rhs) const { return !(rhs == *this); }
+
+	template <class Ar>
+	void serialize(Ar& ar) {
+		serializer(ar, begin, end, result);
+	}
+};
+
+using MappedReqAndResultRef = std::variant<GetValueReqAndResultRef, GetRangeReqAndResultRef>;
+
+struct KeyValueAndMappedReqAndResultRef : KeyValueRef {
+	// Save the original key value at the base (KeyValueRef).
+
+	MappedReqAndResultRef reqAndResult;
+
+	KeyValueAndMappedReqAndResultRef() {}
+	KeyValueAndMappedReqAndResultRef(Arena& a, const KeyValueAndMappedReqAndResultRef& copyFrom)
+	  : KeyValueRef(a, copyFrom) {
+		const auto& reqAndResultCopyFrom = copyFrom.reqAndResult;
+		if (std::holds_alternative<GetValueReqAndResultRef>(reqAndResultCopyFrom)) {
+			auto getValue = std::get<GetValueReqAndResultRef>(reqAndResultCopyFrom);
+			reqAndResult = GetValueReqAndResultRef(a, getValue);
+		} else if (std::holds_alternative<GetRangeReqAndResultRef>(reqAndResultCopyFrom)) {
+			auto getRange = std::get<GetRangeReqAndResultRef>(reqAndResultCopyFrom);
+			reqAndResult = GetRangeReqAndResultRef(a, getRange);
+		} else {
+			throw internal_error();
+		}
+	}
+
+	bool operator==(const KeyValueAndMappedReqAndResultRef& rhs) const {
+		return static_cast<const KeyValueRef&>(*this) == static_cast<const KeyValueRef&>(rhs) &&
+		       reqAndResult == rhs.reqAndResult;
+	}
+	bool operator!=(const KeyValueAndMappedReqAndResultRef& rhs) const { return !(rhs == *this); }
+
+	// It relies on the base to provide the expectedSize. TODO: Consider add the underlying request and key values into
+	// expected size?
+	//	int expectedSize() const { return ((KeyValueRef*)this)->expectedSisze() + reqA }
+
+	template <class Ar>
+	void serialize(Ar& ar) {
+		serializer(ar, ((KeyValueRef&)*this), reqAndResult);
+	}
+};
+
+struct RangeAndMapResultRef : VectorRef<KeyValueAndMappedReqAndResultRef> {
+	// Additional information on range result. See comments on RangeResultRef.
+	bool more;
+	Optional<KeyRef> readThrough;
+	bool readToBegin;
+	bool readThroughEnd;
+
+	RangeAndMapResultRef() : more(false), readToBegin(false), readThroughEnd(false) {}
+	RangeAndMapResultRef(Arena& p, const RangeAndMapResultRef& toCopy)
+	  : VectorRef<KeyValueAndMappedReqAndResultRef>(p, toCopy), more(toCopy.more),
+	    readThrough(toCopy.readThrough.present() ? KeyRef(p, toCopy.readThrough.get()) : Optional<KeyRef>()),
+	    readToBegin(toCopy.readToBegin), readThroughEnd(toCopy.readThroughEnd) {}
+	RangeAndMapResultRef(const VectorRef<KeyValueAndMappedReqAndResultRef>& value,
+	                     bool more,
+	                     Optional<KeyRef> readThrough = Optional<KeyRef>())
+	  : VectorRef<KeyValueAndMappedReqAndResultRef>(value), more(more), readThrough(readThrough), readToBegin(false),
+	    readThroughEnd(false) {}
+	RangeAndMapResultRef(bool readToBegin, bool readThroughEnd)
+	  : more(false), readToBegin(readToBegin), readThroughEnd(readThroughEnd) {}
+
+	template <class Ar>
+	void serialize(Ar& ar) {
+		serializer(
+		    ar, ((VectorRef<KeyValueAndMappedReqAndResultRef>&)*this), more, readThrough, readToBegin, readThroughEnd);
+	}
+
+	std::string toString() const {
+		return "more:" + std::to_string(more) +
+		       " readThrough:" + (readThrough.present() ? readThrough.get().toString() : "[unset]") +
+		       " readToBegin:" + std::to_string(readToBegin) + " readThroughEnd:" + std::to_string(readThroughEnd);
+	}
+};
+
 struct KeyValueStoreType {
 	constexpr static FileIdentifier file_identifier = 6560359;
 	// These enumerated values are stored in the database configuration, so should NEVER be changed.

fdbclient/IClientApi.h

@@ -59,12 +59,12 @@ class ITransaction {
 	                                           GetRangeLimits limits,
 	                                           bool snapshot = false,
 	                                           bool reverse = false) = 0;
-	virtual ThreadFuture<RangeResult> getRangeAndFlatMap(const KeySelectorRef& begin,
-	                                                     const KeySelectorRef& end,
-	                                                     const StringRef& mapper,
-	                                                     GetRangeLimits limits,
-	                                                     bool snapshot = false,
-	                                                     bool reverse = false) = 0;
+	virtual ThreadFuture<RangeAndMapResult> getRangeAndFlatMap(const KeySelectorRef& begin,
+	                                                           const KeySelectorRef& end,
+	                                                           const StringRef& mapper,
+	                                                           GetRangeLimits limits,
+	                                                           bool snapshot = false,
+	                                                           bool reverse = false) = 0;
 	virtual ThreadFuture<Standalone<VectorRef<const char*>>> getAddressesForKey(const KeyRef& key) = 0;
 	virtual ThreadFuture<Standalone<StringRef>> getVersionstamp() = 0;
 

fdbclient/ISingleThreadTransaction.h

@@ -63,12 +63,12 @@ class ISingleThreadTransaction : public ReferenceCounted<ISingleThreadTransactio
 	                                     GetRangeLimits limits,
 	                                     Snapshot = Snapshot::False,
 	                                     Reverse = Reverse::False) = 0;
-	virtual Future<RangeResult> getRangeAndFlatMap(KeySelector begin,
-	                                               KeySelector end,
-	                                               Key mapper,
-	                                               GetRangeLimits limits,
-	                                               Snapshot = Snapshot::False,
-	                                               Reverse = Reverse::False) = 0;
+	virtual Future<RangeAndMapResult> getRangeAndFlatMap(KeySelector begin,
+	                                                     KeySelector end,
+	                                                     Key mapper,
+	                                                     GetRangeLimits limits,
+	                                                     Snapshot = Snapshot::False,
+	                                                     Reverse = Reverse::False) = 0;
 	virtual Future<Standalone<VectorRef<const char*>>> getAddressesForKey(Key const& key) = 0;
 	virtual Future<Standalone<VectorRef<KeyRef>>> getRangeSplitPoints(KeyRange const& range, int64_t chunkSize) = 0;
 	virtual Future<int64_t> getEstimatedRangeSizeBytes(KeyRange const& keys) = 0;