From 82febbee63b08351e3f6b89538bbf96b1ec1f4b4 Mon Sep 17 00:00:00 2001 From: Ronit Sabhaya Date: Mon, 19 Oct 2026 18:10:12 -0500 Subject: [PATCH 1/2] NativeAPI: Update getRangeStream read conflict ranges continuously as results are returned --- fdbclient/NativeAPI.cpp | 48 +++++++++++++++++-------- fdbclient/include/fdbclient/NativeAPI.h | 15 +++++++- 2 files changed, 47 insertions(+), 16 deletions(-) diff --git a/fdbclient/NativeAPI.cpp b/fdbclient/NativeAPI.cpp index 34d7eeb15fe..e8f72f4d273 100644 --- a/fdbclient/NativeAPI.cpp +++ b/fdbclient/NativeAPI.cpp @@ -3181,6 +3181,7 @@ Optional> maybeDuplicateTSSStr Future getRangeStreamImpl(Reference trStateInput, PromiseStream resultsInput, KeyRange keysInput, + KeyRange conflictBounds, GetRangeLimits limits, Snapshot snapshot, Reverse reverse, @@ -3188,6 +3189,7 @@ Future getRangeStreamImpl(Reference trStateInput, Reference trState(std::move(trStateInput)); PromiseStream results(std::move(resultsInput)); KeyRange keys(std::move(keysInput)); + Key currentConflictBegin = conflictBounds.begin; std::vector locations; // Keep the active streams alive through retry delays and location refreshes. Optional> tssDuplicateStream; @@ -3401,6 +3403,13 @@ Future getRangeStreamImpl(Reference 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()) { @@ -3417,6 +3426,13 @@ Future getRangeStreamImpl(Reference trStateInput, // 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); + if (!snapshot) { + Key chunkEnd = reverse ? range.begin : range.end; + if (currentConflictBegin < chunkEnd) { + trState->addStreamConflictRange(KeyRangeRef(currentConflictBegin, chunkEnd)); + currentConflictBegin = chunkEnd; + } + } results.send(std::move(output)); break; } @@ -3428,6 +3444,13 @@ Future getRangeStreamImpl(Reference trStateInput, if (keys.end == allKeys.end && reverse) { output.readThroughEnd = true; } + if (!snapshot) { + Key chunkEnd = reverse ? output[0].key : keyAfter(output[output.size() - 1].key); + if (currentConflictBegin < chunkEnd) { + trState->addStreamConflictRange(KeyRangeRef(currentConflictBegin, chunkEnd)); + currentConflictBegin = chunkEnd; + } + } results.send(std::move(output)); } if (breakAgain) { @@ -3473,14 +3496,12 @@ Future getRangeStream(Reference trStateInput, KeySelector beginInput, KeySelector endInput, GetRangeLimits limits, - Promise> conflictRangeInput, Snapshot snapshot, Reverse reverse) { Reference trState(std::move(trStateInput)); PromiseStream _results(std::move(_resultsInput)); KeySelector begin(std::move(beginInput)); KeySelector end(std::move(endInput)); - Promise> conflictRange(std::move(conflictRangeInput)); // FIXME: better handling to disable row limits ASSERT(!limits.hasRowLimit()); Span span("NAPI:getRangeStream"_loc, trState->spanContext); @@ -3495,19 +3516,16 @@ Future getRangeStream(Reference 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) { _results.sendError(end_of_stream()); co_return; } + Key rangeBegin = std::min(b, Key(begin.getKey(), begin.arena())); + Key rangeEnd = std::max(e, Key(end.getKey(), end.arena())); + co_await getRangeStreamImpl( - trState, _results, KeyRange(KeyRangeRef(b, e)), limits, snapshot, reverse, span.context); + trState, _results, KeyRange(KeyRangeRef(b, e)), KeyRange(KeyRangeRef(rangeBegin, rangeEnd)), limits, snapshot, reverse, span.context); } Future getRange(Reference const& trState, @@ -3940,12 +3958,7 @@ Future Transaction::getRangeStream(PromiseStream& results, return Void(); } - Promise> 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 Transaction::getRangeStream(PromiseStream& results, @@ -4840,6 +4853,11 @@ Future Transaction::commitMutations() { tr.transaction.read_conflict_ranges.emplace_back( tr.arena, extraConflictRange.get().first, extraConflictRange.get().second); } + for (const auto& streamRange : trState->streamConflictRanges) { + if (streamRange.begin < streamRange.end) + tr.transaction.read_conflict_ranges.emplace_back( + tr.arena, streamRange.begin, streamRange.end); + } if (tr.idempotencyId.valid()) { // We need to be able confirm that this transaction is no longer in diff --git a/fdbclient/include/fdbclient/NativeAPI.h b/fdbclient/include/fdbclient/NativeAPI.h index 33a9775818c..2d7e2297d98 100644 --- a/fdbclient/include/fdbclient/NativeAPI.h +++ b/fdbclient/include/fdbclient/NativeAPI.h @@ -276,6 +276,13 @@ struct TransactionState : ReferenceCounted { bool automaticIdempotency = false; + std::vector streamConflictRanges; + void addStreamConflictRange(KeyRangeRef const& range) { + if (range.begin < range.end) { + streamConflictRanges.emplace_back(range); + } + } + Future startFuture; // Only available so that Transaction can have a default constructor, for use in state variables @@ -484,7 +491,13 @@ class Transaction : NonCopyable { const std::vector>>& getExtraReadConflictRanges() const { return extraConflictRanges; } Standalone> readConflictRanges() const { - return Standalone>(tr.transaction.read_conflict_ranges, tr.arena); + Standalone> result(tr.transaction.read_conflict_ranges, tr.arena); + for (const auto& r : trState->streamConflictRanges) { + if (r.begin < r.end) { + result.push_back(result.arena(), r); + } + } + return result; } Standalone> writeConflictRanges() const { return Standalone>(tr.transaction.write_conflict_ranges, tr.arena); From b783367acb8a0a5d9780b113b0eb5b50b00e2302 Mon Sep 17 00:00:00 2001 From: Ronit Sabhaya Date: Wed, 21 Oct 2026 20:08:34 -0500 Subject: [PATCH 2/2] updating the nativeapi with changes saintstack suggested --- fdbclient/NativeAPI.cpp | 212 ++++++++++++++++++++++-- fdbclient/include/fdbclient/NativeAPI.h | 24 ++- 2 files changed, 219 insertions(+), 17 deletions(-) diff --git a/fdbclient/NativeAPI.cpp b/fdbclient/NativeAPI.cpp index e8f72f4d273..e11656b84ed 100644 --- a/fdbclient/NativeAPI.cpp +++ b/fdbclient/NativeAPI.cpp @@ -3189,6 +3189,7 @@ Future getRangeStreamImpl(Reference trStateInput, Reference trState(std::move(trStateInput)); PromiseStream results(std::move(resultsInput)); KeyRange keys(std::move(keysInput)); + ASSERT(!reverse); Key currentConflictBegin = conflictBounds.begin; std::vector locations; // Keep the active streams alive through retry delays and location refreshes. @@ -3425,9 +3426,9 @@ Future getRangeStreamImpl(Reference 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 = reverse ? range.begin : range.end; + Key chunkEnd = range.end; if (currentConflictBegin < chunkEnd) { trState->addStreamConflictRange(KeyRangeRef(currentConflictBegin, chunkEnd)); currentConflictBegin = chunkEnd; @@ -3445,7 +3446,7 @@ Future getRangeStreamImpl(Reference trStateInput, output.readThroughEnd = true; } if (!snapshot) { - Key chunkEnd = reverse ? output[0].key : keyAfter(output[output.size() - 1].key); + Key chunkEnd = keyAfter(output[output.size() - 1].key); if (currentConflictBegin < chunkEnd) { trState->addStreamConflictRange(KeyRangeRef(currentConflictBegin, chunkEnd)); currentConflictBegin = chunkEnd; @@ -3490,6 +3491,36 @@ Future>> getRangeSplitPoints(Reference= 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 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 getRangeStream(Reference trStateInput, PromiseStream _resultsInput, @@ -3503,7 +3534,7 @@ Future getRangeStream(Reference trStateInput, KeySelector begin(std::move(beginInput)); KeySelector end(std::move(endInput)); // 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(); @@ -3516,16 +3547,14 @@ Future getRangeStream(Reference trStateInput, Key b = co_await fb; Key e = co_await fe; - if (b >= e) { + KeyRange conflictBounds; + if (recordRangeStreamSelectorConflicts(trState, b, e, begin, end, snapshot, conflictBounds)) { _results.sendError(end_of_stream()); co_return; } - Key rangeBegin = std::min(b, Key(begin.getKey(), begin.arena())); - Key rangeEnd = std::max(e, Key(end.getKey(), end.arena())); - co_await getRangeStreamImpl( - trState, _results, KeyRange(KeyRangeRef(b, e)), KeyRange(KeyRangeRef(rangeBegin, rangeEnd)), limits, snapshot, reverse, span.context); + trState, _results, KeyRange(KeyRangeRef(b, e)), conflictBounds, limits, snapshot, reverse, span.context); } Future getRange(Reference const& trState, @@ -4826,6 +4855,11 @@ Future 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") @@ -4853,11 +4887,6 @@ Future Transaction::commitMutations() { tr.transaction.read_conflict_ranges.emplace_back( tr.arena, extraConflictRange.get().first, extraConflictRange.get().second); } - for (const auto& streamRange : trState->streamConflictRanges) { - if (streamRange.begin < streamRange.end) - tr.transaction.read_conflict_ranges.emplace_back( - tr.arena, streamRange.begin, streamRange.end); - } if (tr.idempotencyId.valid()) { // We need to be able confirm that this transaction is no longer in @@ -6538,6 +6567,161 @@ TEST_CASE("/fdbclient/NativeAPI/rangeSplitPoints/multipleShards") { return Void(); } +TEST_CASE("/fdbclient/NativeAPI/streamConflictRanges/mergeContiguousAndOverlapping") { + auto trState = makeReference(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> 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(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(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(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(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>(); + Database cx = DatabaseContext::create(dbInfo, Future(), 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>(); + Database cx = DatabaseContext::create(dbInfo, Future(), 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 setPerpetualStorageWiggle(Database cx, bool enable, LockAware lockAware) { ReadYourWritesTransaction tr(cx); while (true) { diff --git a/fdbclient/include/fdbclient/NativeAPI.h b/fdbclient/include/fdbclient/NativeAPI.h index 2d7e2297d98..915302afcc6 100644 --- a/fdbclient/include/fdbclient/NativeAPI.h +++ b/fdbclient/include/fdbclient/NativeAPI.h @@ -278,9 +278,27 @@ struct TransactionState : ReferenceCounted { std::vector streamConflictRanges; void addStreamConflictRange(KeyRangeRef const& range) { - if (range.begin < range.end) { - streamConflictRanges.emplace_back(range); + if (range.begin >= range.end) { + return; + } + Key newBegin(range.begin); + Key newEnd(range.end); + std::vector 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 startFuture; @@ -494,7 +512,7 @@ class Transaction : NonCopyable { Standalone> result(tr.transaction.read_conflict_ranges, tr.arena); for (const auto& r : trState->streamConflictRanges) { if (r.begin < r.end) { - result.push_back(result.arena(), r); + result.push_back(result.arena(), KeyRangeRef(result.arena(), r)); } } return result;