Skip to content

Commit

Permalink
Change API to include all primary results, WIP
Browse files Browse the repository at this point in the history
  • Loading branch information
@nblintao
nblintao committed Jan 11, 2022
1 parent 57e6bf0 commit cc4ec67
Show file tree
Hide file tree
Showing 14 changed files with 302 additions and 178 deletions.
1 change: 1 addition & 0 deletions bindings/c/fdb_c.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -570,6 +570,7 @@ FDBFuture* fdb_transaction_get_range_impl(FDBTransaction* tr,
.extractPtr());
}

// TODO: Interpret FDBFuture as RangeAndMapResult.
FDBFuture* fdb_transaction_get_range_and_flat_map_impl(FDBTransaction* tr,
uint8_t const* begin_key_name,
int begin_key_name_length,
Expand Down
82 changes: 74 additions & 8 deletions fdbclient/FDBTypes.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -611,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 };

Expand All @@ -624,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();
Expand Down Expand Up @@ -680,9 +684,17 @@ 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) {}

int expectedSize() const { return key.expectedSize() + result.expectedSize(); }

template <class Ar>
void serialize(Ar& ar) {
serializer(ar, key, result);
Expand All @@ -695,9 +707,10 @@ struct GetRangeReqAndResultRef {
//
// KeyRef key;
// ValueRef value;
// KeyValueRef() {}
GetRangeReqAndResultRef() {}
// KeyValueRef(const KeyRef& key, const ValueRef& value) : key(key), value(value) {}
// KeyValueRef(Arena& a, const KeyValueRef& copyFrom) : key(a, copyFrom.key), value(a, copyFrom.value) {}
GetRangeReqAndResultRef(Arena& a, const GetRangeReqAndResultRef& copyFrom)
: begin(a, copyFrom.begin), end(a, copyFrom.end), result(a, copyFrom.result) {}
// bool operator==(const KeyValueRef& r) const { return key == r.key && value == r.value; }
// bool operator!=(const KeyValueRef& r) const { return key != r.key || value != r.value; }

Expand All @@ -709,14 +722,67 @@ struct GetRangeReqAndResultRef {

using ReqAndResultRef = std::variant<GetValueReqAndResultRef, GetRangeReqAndResultRef>;

struct RangeAndMapResultRef : VectorRef<ReqAndResultRef> {
RangeResultRef originalRangeResult; // In addition to metadata (including more, readToBegin, and readThroughEnd), it
// only stores the first and the last result. This is useful for the callers to
// know how far it has reached.
RangeAndMapResultRef() {}
struct KeyValueAndMappedReqAndResultRef : KeyValueRef {
// Save the original key value at the base (KeyValueRef).

ReqAndResultRef 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& r) const = delete; // TODO
bool operator!=(const KeyValueAndMappedReqAndResultRef& r) const = delete; // TODO

// 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, ((VectorRef<ReqAndResultRef>&)*this), originalRangeResult);
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);
}
};

Expand Down
12 changes: 6 additions & 6 deletions fdbclient/IClientApi.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down
12 changes: 6 additions & 6 deletions fdbclient/ISingleThreadTransaction.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
12 changes: 6 additions & 6 deletions fdbclient/MultiVersionTransaction.actor.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -861,12 +861,12 @@ ThreadFuture<RangeResult> MultiVersionTransaction::getRange(const KeyRangeRef& k
return abortableFuture(f, tr.onChange);
}

ThreadFuture<RangeResult> MultiVersionTransaction::getRangeAndFlatMap(const KeySelectorRef& begin,
const KeySelectorRef& end,
const StringRef& mapper,
GetRangeLimits limits,
bool snapshot,
bool reverse) {
ThreadFuture<RangeAndMapResult> MultiVersionTransaction::getRangeAndFlatMap(const KeySelectorRef& begin,
const KeySelectorRef& end,
const StringRef& mapper,
GetRangeLimits limits,
bool snapshot,
bool reverse) {
auto tr = getTransaction();
auto f = tr.transaction ? tr.transaction->getRangeAndFlatMap(begin, end, mapper, limits, snapshot, reverse)
: makeTimeout<RangeResult>();
Expand Down
12 changes: 6 additions & 6 deletions fdbclient/MultiVersionTransaction.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -434,12 +434,12 @@ class MultiVersionTransaction : public ITransaction, ThreadSafeReferenceCounted<
GetRangeLimits limits,
bool snapshot = false,
bool reverse = false) override;
ThreadFuture<RangeResult> getRangeAndFlatMap(const KeySelectorRef& begin,
const KeySelectorRef& end,
const StringRef& mapper,
GetRangeLimits limits,
bool snapshot,
bool reverse) override;
ThreadFuture<RangeAndMapResult> getRangeAndFlatMap(const KeySelectorRef& begin,
const KeySelectorRef& end,
const StringRef& mapper,
GetRangeLimits limits,
bool snapshot,
bool reverse) override;
ThreadFuture<Standalone<VectorRef<const char*>>> getAddressesForKey(const KeyRef& key) override;
ThreadFuture<Standalone<StringRef>> getVersionstamp() override;

Expand Down
Loading

0 comments on commit cc4ec67

Please sign in to comment.