Skip to content

NativeAPI: Update getRangeStream read conflict ranges continuously as results are returned #14323

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Ronitsabhaya75 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

NativeAPI: Update getRangeStream read conflict ranges continuously as results are returned #14323

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
82febbe
NativeAPI: Update getRangeStream read conflict ranges continuously as…
Ronitsabhaya75 Oct 19, 2026
b783367
updating the nativeapi with changes saintstack suggested
Ronitsabhaya75 Oct 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
238 changes: 220 additions & 18 deletions fdbclient/NativeAPI.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3181,13 +3181,16 @@ Optional<TSSDuplicateStreamData<REPLYSTREAM_TYPE(Request)>> maybeDuplicateTSSStr
Future<Void> getRangeStreamImpl(Reference<TransactionState> trStateInput,
PromiseStream<RangeResult> resultsInput,
KeyRange keysInput,
KeyRange conflictBounds,
GetRangeLimits limits,
Snapshot snapshot,
Reverse reverse,
SpanContext spanContext) {
Reference<TransactionState> trState(std::move(trStateInput));
PromiseStream<RangeResult> results(std::move(resultsInput));
KeyRange keys(std::move(keysInput));
ASSERT(!reverse);
Key currentConflictBegin = conflictBounds.begin;
std::vector<KeyRangeLocationInfo> locations;
// Keep the active streams alive through retry delays and location refreshes.
Optional<TSSDuplicateStreamData<GetKeyValuesStreamReply>> tssDuplicateStream;
Expand Down Expand Up @@ -3401,6 +3404,13 @@ Future<Void> getRangeStreamImpl(Reference<TransactionState> trStateInput,
// getRangeStream() uses end_of_stream to indicate exhaustion, so keep 'more' true.
output.more = true;
output.setReadThrough(reverse ? keys.begin : keys.end);
if (!snapshot) {
Key chunkEnd = conflictBounds.end;
if (currentConflictBegin < chunkEnd) {
trState->addStreamConflictRange(KeyRangeRef(currentConflictBegin, chunkEnd));
currentConflictBegin = chunkEnd;
}
}
results.send(std::move(output));
results.sendError(end_of_stream());
if (tssDuplicateStream.present() && !tssDuplicateStream.get().done()) {
Expand All @@ -3416,7 +3426,14 @@ Future<Void> getRangeStreamImpl(Reference<TransactionState> trStateInput,
output.arena().dependsOn(range.arena());
// If it's not the last shard, set more to true and readThrough to the shard boundary.
output.more = true;
output.setReadThrough(reverse ? range.begin : range.end);
output.setReadThrough(range.end);
if (!snapshot) {
Key chunkEnd = range.end;
if (currentConflictBegin < chunkEnd) {
trState->addStreamConflictRange(KeyRangeRef(currentConflictBegin, chunkEnd));
currentConflictBegin = chunkEnd;
}
}
results.send(std::move(output));
break;
}
Expand All @@ -3428,6 +3445,13 @@ Future<Void> getRangeStreamImpl(Reference<TransactionState> trStateInput,
if (keys.end == allKeys.end && reverse) {
output.readThroughEnd = true;
}
if (!snapshot) {
Key chunkEnd = keyAfter(output[output.size() - 1].key);
if (currentConflictBegin < chunkEnd) {
trState->addStreamConflictRange(KeyRangeRef(currentConflictBegin, chunkEnd));
currentConflictBegin = chunkEnd;
}
}
results.send(std::move(output));
}
if (breakAgain) {
Expand Down Expand Up @@ -3467,22 +3491,50 @@ Future<Standalone<VectorRef<KeyRef>>> getRangeSplitPoints(Reference<TransactionS
KeyRange keys,
int64_t chunkSize,
int limit);
// Evaluates selector bounds and records conflict ranges for:
// - Empty resolved ranges (b >= e) spanning [rangeBegin, rangeEnd)
// - Resolved end selectors with end.offset <= 0 spanning [e, rangeEnd)
// Returns true if the range is empty (b >= e) and streaming should terminate.
bool recordRangeStreamSelectorConflicts(Reference<TransactionState> const& trState,
Key const& b,
Key const& e,
KeySelector const& begin,
KeySelector const& end,
Snapshot snapshot,
KeyRange& conflictBounds) {
Key rangeBegin = std::min(b, Key(begin.getKey(), begin.arena()));
Key rangeEnd = std::max(e, Key(end.getKey(), end.arena()));

if (b >= e) {
if (!snapshot && rangeBegin < rangeEnd) {
trState->addStreamConflictRange(KeyRangeRef(rangeBegin, rangeEnd));
}
return true;
}

conflictBounds = KeyRange(KeyRangeRef(rangeBegin, rangeEnd));

if (!snapshot && end.offset <= 0 && e < rangeEnd) {
trState->addStreamConflictRange(KeyRangeRef(e, rangeEnd));
}

return false;
}

// Streams the requested key range directly from storage servers without fragment-level parallelism.
Future<Void> getRangeStream(Reference<TransactionState> trStateInput,
PromiseStream<RangeResult> _resultsInput,
KeySelector beginInput,
KeySelector endInput,
GetRangeLimits limits,
Promise<std::pair<Key, Key>> conflictRangeInput,
Snapshot snapshot,
Reverse reverse) {
Reference<TransactionState> trState(std::move(trStateInput));
PromiseStream<RangeResult> _results(std::move(_resultsInput));
KeySelector begin(std::move(beginInput));
KeySelector end(std::move(endInput));
Promise<std::pair<Key, Key>> conflictRange(std::move(conflictRangeInput));
// FIXME: better handling to disable row limits
ASSERT(!limits.hasRowLimit());
ASSERT(!limits.hasRowLimit() && !reverse);
Span span("NAPI:getRangeStream"_loc, trState->spanContext);

co_await trState->startTransaction();
Expand All @@ -3495,19 +3547,14 @@ Future<Void> getRangeStream(Reference<TransactionState> trStateInput,
Key b = co_await fb;
Key e = co_await fe;

if (!snapshot) {
// FIXME: this conflict range is too large, and should be updated continuously as results are returned
conflictRange.send(std::make_pair(std::min(b, Key(begin.getKey(), begin.arena())),
std::max(e, Key(end.getKey(), end.arena()))));
}

if (b >= e) {
KeyRange conflictBounds;
if (recordRangeStreamSelectorConflicts(trState, b, e, begin, end, snapshot, conflictBounds)) {
_results.sendError(end_of_stream());
co_return;
}

co_await getRangeStreamImpl(
trState, _results, KeyRange(KeyRangeRef(b, e)), limits, snapshot, reverse, span.context);
trState, _results, KeyRange(KeyRangeRef(b, e)), conflictBounds, limits, snapshot, reverse, span.context);
}

Future<RangeResult> getRange(Reference<TransactionState> const& trState,
Expand Down Expand Up @@ -3940,12 +3987,7 @@ Future<Void> Transaction::getRangeStream(PromiseStream<RangeResult>& results,
return Void();
}

Promise<std::pair<Key, Key>> conflictRange;
if (!snapshot) {
extraConflictRanges.push_back(conflictRange.getFuture());
}

return forwardErrors(::getRangeStream(trState, results, b, e, limits, conflictRange, snapshot, reverse), results);
return forwardErrors(::getRangeStream(trState, results, b, e, limits, snapshot, reverse), results);
}

Future<Void> Transaction::getRangeStream(PromiseStream<RangeResult>& results,
Expand Down Expand Up @@ -4813,6 +4855,11 @@ Future<Void> Transaction::commitMutations() {
if (trState->options.tags.size())
tr.tagSet = trState->options.tags;

for (const auto& streamRange : trState->streamConflictRanges) {
if (streamRange.begin < streamRange.end)
tr.transaction.read_conflict_ranges.emplace_back(tr.arena, streamRange.begin, streamRange.end);
}

size_t transactionSize = getSize();
if (transactionSize > (uint64_t)FLOW_KNOBS->PACKET_WARNING) {
TraceEvent(SevWarn, "LargeTransaction")
Expand Down Expand Up @@ -6520,6 +6567,161 @@ TEST_CASE("/fdbclient/NativeAPI/rangeSplitPoints/multipleShards") {
return Void();
}

TEST_CASE("/fdbclient/NativeAPI/streamConflictRanges/mergeContiguousAndOverlapping") {
auto trState = makeReference<TransactionState>(TaskPriority::DefaultEndpoint, SpanContext());
ASSERT(trState->streamConflictRanges.empty());

// Contiguous ranges should merge into a single range
trState->addStreamConflictRange(KeyRangeRef("a"_sr, "c"_sr));
ASSERT_EQ(trState->streamConflictRanges.size(), 1);
ASSERT(trState->streamConflictRanges[0] == KeyRangeRef("a"_sr, "c"_sr));

trState->addStreamConflictRange(KeyRangeRef("c"_sr, "e"_sr));
ASSERT_EQ(trState->streamConflictRanges.size(), 1);
ASSERT(trState->streamConflictRanges[0] == KeyRangeRef("a"_sr, "e"_sr));

// Overlapping range should expand bounds
trState->addStreamConflictRange(KeyRangeRef("d"_sr, "g"_sr));
ASSERT_EQ(trState->streamConflictRanges.size(), 1);
ASSERT(trState->streamConflictRanges[0] == KeyRangeRef("a"_sr, "g"_sr));

// Disjoint range should be kept separate
trState->addStreamConflictRange(KeyRangeRef("m"_sr, "p"_sr));
ASSERT_EQ(trState->streamConflictRanges.size(), 2);

// Bridging range should merge all touching ranges
trState->addStreamConflictRange(KeyRangeRef("g"_sr, "n"_sr));
ASSERT_EQ(trState->streamConflictRanges.size(), 1);
ASSERT(trState->streamConflictRanges[0] == KeyRangeRef("a"_sr, "p"_sr));

// Empty range should be ignored
trState->addStreamConflictRange(KeyRangeRef("z"_sr, "z"_sr));
ASSERT_EQ(trState->streamConflictRanges.size(), 1);

return Void();
}

TEST_CASE("/fdbclient/NativeAPI/readConflictRanges/deepCopy") {
Transaction tr;
tr.trState->addStreamConflictRange(KeyRangeRef("bar"_sr, "foo"_sr));

Standalone<VectorRef<KeyRangeRef>> ranges = tr.readConflictRanges();
ASSERT_EQ(ranges.size(), 1);
ASSERT(ranges[0] == KeyRangeRef("bar"_sr, "foo"_sr));

// Modifying trState's stream ranges should not affect the returned Standalone result
tr.trState->streamConflictRanges.clear();
ASSERT_EQ(ranges.size(), 1);
ASSERT(ranges[0] == KeyRangeRef("bar"_sr, "foo"_sr));

return Void();
}

TEST_CASE("/fdbclient/NativeAPI/streamConflictRanges/emptyResolvedRange") {
auto trState = makeReference<TransactionState>(TaskPriority::DefaultEndpoint, SpanContext());

Key b = "z"_sr;
Key e = "a"_sr;
KeySelector begin("b"_sr, false, 1);
KeySelector end("y"_sr, false, 0);

KeyRange conflictBounds;
// Non-snapshot empty resolved range (b >= e) records [rangeBegin, rangeEnd) == ["b", "y")
bool isEmpty = recordRangeStreamSelectorConflicts(trState, b, e, begin, end, Snapshot::False, conflictBounds);
ASSERT(isEmpty);
ASSERT_EQ(trState->streamConflictRanges.size(), 1);
ASSERT(trState->streamConflictRanges[0] == KeyRangeRef("b"_sr, "y"_sr));

// Snapshot read records no conflict range
auto trStateSnapshot = makeReference<TransactionState>(TaskPriority::DefaultEndpoint, SpanContext());
isEmpty = recordRangeStreamSelectorConflicts(trStateSnapshot, b, e, begin, end, Snapshot::True, conflictBounds);
ASSERT(isEmpty);
ASSERT(trStateSnapshot->streamConflictRanges.empty());

return Void();
}

TEST_CASE("/fdbclient/NativeAPI/streamConflictRanges/partialConsumptionAndEndSelector") {
auto trState = makeReference<TransactionState>(TaskPriority::DefaultEndpoint, SpanContext());

Key b = "c"_sr;
Key e = "v"_sr;
KeySelector begin("c"_sr, false, 1);
KeySelector end("y"_sr, false, 0); // end.offset <= 0

KeyRange conflictBounds;
bool isEmpty = recordRangeStreamSelectorConflicts(trState, b, e, begin, end, Snapshot::False, conflictBounds);
ASSERT(!isEmpty);
ASSERT(conflictBounds == KeyRangeRef("c"_sr, "y"_sr));

// Production helper evaluated end.offset <= 0 and registered [e, rangeEnd) == ["v", "y")
ASSERT_EQ(trState->streamConflictRanges.size(), 1);
ASSERT(trState->streamConflictRanges[0] == KeyRangeRef("v"_sr, "y"_sr));

// Partial consumption: client consumes chunk 1 ["c", "f") and abandons the stream
trState->addStreamConflictRange(KeyRangeRef(conflictBounds.begin, "f"_sr));

// Two disjoint ranges: consumed chunk ["c", "f") and selector ["v", "y").
// Unread middle interval ["f", "v") has no conflict range, preventing false aborts.
ASSERT_EQ(trState->streamConflictRanges.size(), 2);

// If the stream is consumed to completion, subsequent chunks ["f", "y") coalesce all ranges into ["c", "y")
trState->addStreamConflictRange(KeyRangeRef("f"_sr, conflictBounds.end));
ASSERT_EQ(trState->streamConflictRanges.size(), 1);
ASSERT(trState->streamConflictRanges[0] == KeyRangeRef("c"_sr, "y"_sr));

// For end selector with offset > 0 (e.g. firstGreaterOrEqual), no selector range is registered upfront
auto trStateForward = makeReference<TransactionState>(TaskPriority::DefaultEndpoint, SpanContext());
KeySelector endForward("y"_sr, false, 1);
isEmpty =
recordRangeStreamSelectorConflicts(trStateForward, b, e, begin, endForward, Snapshot::False, conflictBounds);
ASSERT(!isEmpty);
ASSERT(trStateForward->streamConflictRanges.empty());

return Void();
}

TEST_CASE("/fdbclient/NativeAPI/streamConflictRanges/resetClearsConflicts") {
auto dbInfo = makeReference<AsyncVar<ClientDBInfo>>();
Database cx = DatabaseContext::create(dbInfo, Future<Void>(), LocalityData(), EnableLocalityLoadBalance::False);
Transaction tr(cx);

tr.trState->addStreamConflictRange(KeyRangeRef("a"_sr, "b"_sr));
ASSERT_EQ(tr.readConflictRanges().size(), 1);

tr.reset();
ASSERT(tr.readConflictRanges().empty());

tr.trState->addStreamConflictRange(KeyRangeRef("c"_sr, "d"_sr));
ASSERT_EQ(tr.readConflictRanges().size(), 1);

tr.fullReset();
ASSERT(tr.readConflictRanges().empty());

return Void();
}

TEST_CASE("/fdbclient/NativeAPI/streamConflictRanges/commitSizeCalculation") {
auto dbInfo = makeReference<AsyncVar<ClientDBInfo>>();
Database cx = DatabaseContext::create(dbInfo, Future<Void>(), LocalityData(), EnableLocalityLoadBalance::False);
Transaction tr(cx);

// Add a mutation so commitMutations does not early-exit as read-only
tr.set("foo"_sr, "bar"_sr);
tr.trState->addStreamConflictRange(KeyRangeRef("a"_sr, "z"_sr));

uint32_t sizeBefore = tr.getSize();

// Enforce size limit equal to sizeBefore; any added stream conflict bytes guarantee exceeding it
tr.trState->options.sizeLimit = sizeBefore;

auto commitFuture = tr.commitMutations();
ASSERT(commitFuture.isError());
ASSERT_EQ(commitFuture.getError().code(), error_code_transaction_too_large);

return Void();
}

Future<Version> setPerpetualStorageWiggle(Database cx, bool enable, LockAware lockAware) {
ReadYourWritesTransaction tr(cx);
while (true) {
Expand Down
33 changes: 32 additions & 1 deletion fdbclient/include/fdbclient/NativeAPI.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -276,6 +276,31 @@ struct TransactionState : ReferenceCounted<TransactionState> {

bool automaticIdempotency = false;

std::vector<KeyRange> streamConflictRanges;
void addStreamConflictRange(KeyRangeRef const& range) {
if (range.begin >= range.end) {
return;
}
Key newBegin(range.begin);
Key newEnd(range.end);
std::vector<KeyRange> merged;
merged.reserve(streamConflictRanges.size() + 1);
for (const auto& existing : streamConflictRanges) {
if (existing.end >= newBegin && newEnd >= existing.begin) {
if (existing.begin < newBegin) {
newBegin = existing.begin;
}
if (existing.end > newEnd) {
newEnd = existing.end;
}
} else {
merged.push_back(existing);
}
}
merged.push_back(KeyRange(KeyRangeRef(newBegin, newEnd)));
streamConflictRanges = std::move(merged);
}

Future<Void> startFuture;

// Only available so that Transaction can have a default constructor, for use in state variables
Expand Down Expand Up @@ -484,7 +509,13 @@ class Transaction : NonCopyable {

const std::vector<Future<std::pair<Key, Key>>>& getExtraReadConflictRanges() const { return extraConflictRanges; }
Standalone<VectorRef<KeyRangeRef>> readConflictRanges() const {
return Standalone<VectorRef<KeyRangeRef>>(tr.transaction.read_conflict_ranges, tr.arena);
Standalone<VectorRef<KeyRangeRef>> result(tr.transaction.read_conflict_ranges, tr.arena);
for (const auto& r : trState->streamConflictRanges) {
if (r.begin < r.end) {
result.push_back(result.arena(), KeyRangeRef(result.arena(), r));
}
}
return result;
}
Standalone<VectorRef<KeyRangeRef>> writeConflictRanges() const {
return Standalone<VectorRef<KeyRangeRef>>(tr.transaction.write_conflict_ranges, tr.arena);
Expand Down