db: streams hiding pagination in kv::api::Transaction range Data API (#2864)

db: streams hiding pagination in kv::api::Transaction range Data API
db: implement direct kv::api::LocalTransaction using ranges/views
db: move paginated implementation to kv::api::RemoteTransaction
db: support skipping empty values in range queries for direct implementation
db: fix empty end key in range queries
db: move ranges' vector from iterator to view in MergeManyView
rpc: fix block parameter lifecycle in parity_listStorageKeys
rpc: rename PaginatedStream as Stream

Author: canepat

.github/workflows/run_integration_tests.sh

@@ -15,8 +15,6 @@ rm -rf ./mainnet/results/
 # eth_getLogs: waiting erigon fix on wrong FirstLogIndex in ReceiptsDomain
 # debug_traceBlockByNumber[24-28]: response different wrt erigon
 python3 ./run_tests.py --continue --blockchain mainnet --jwt "$2" --display-only-fail --json-diff --port 51515 --transport_type http -x \
-debug_accountRange,\
-debug_storageRangeAt,\
 debug_traceBlockByNumber/test_24,\
 debug_traceBlockByNumber/test_25,\
 debug_traceBlockByNumber/test_26,\
@@ -36,7 +34,6 @@ eth_getLogs/test_17,\
 eth_getLogs/test_18,\
 eth_getLogs/test_19,\
 eth_getLogs/test_20,\
-parity_listStorageKeys,\
 trace_replayBlockTransactions/test_29,\
 trace_transaction/test_44,\
 trace_transaction/test_47

silkworm/db/datastore/common/ranges/merge_many_view.hpp

@@ -43,13 +43,10 @@ class MergeManyView : public std::ranges::view_interface<MergeManyView<Range, Ra
         using pointer = std::remove_reference_t<reference>*;
 
         Iterator() = default;
-        Iterator(
-            Ranges& ranges,
-            const Comp* comp, Proj proj)
-            : ranges_{vector_from_range(ranges)},
-              comp_{comp},
+        Iterator(std::vector<Range>& ranges, const Comp* comp, Proj proj)
+            : comp_{comp},
               proj_{std::move(proj)} {
-            for (Range& range : ranges_) {
+            for (Range& range : ranges) {
                 iterators_.emplace_back(std::ranges::begin(range));
                 sentinels_.emplace_back(std::ranges::end(range));
             }
@@ -156,7 +153,6 @@ class MergeManyView : public std::ranges::view_interface<MergeManyView<Range, Ra
             return std::ranges::iter_move(iterators_[i]);
         }
 
-        std::vector<Range> ranges_;
         std::vector<RangeIterator> iterators_;
         std::vector<RangeSentinel> sentinels_;
         const Comp* comp_{nullptr};
@@ -170,7 +166,7 @@ class MergeManyView : public std::ranges::view_interface<MergeManyView<Range, Ra
     MergeManyView(
         Ranges ranges,
         Comp comp, Proj proj)
-        : ranges_{std::move(ranges)},
+        : ranges_{vector_from_range(std::move(ranges))},
           comp_{std::move(comp)},
           proj_{std::move(proj)} {}
     MergeManyView() = default;
@@ -182,7 +178,7 @@ class MergeManyView : public std::ranges::view_interface<MergeManyView<Range, Ra
     std::default_sentinel_t end() const { return std::default_sentinel; }
 
   private:
-    Ranges ranges_;
+    std::vector<Range> ranges_;
     Comp comp_;
     Proj proj_;
 };

silkworm/db/datastore/common/ranges/merge_many_view_test.cpp

@@ -12,6 +12,7 @@ namespace silkworm::views {
 
 static_assert(std::ranges::input_range<MergeManyView<std::vector<std::vector<int>>>>);
 static_assert(std::ranges::view<MergeManyView<std::vector<std::vector<int>>>>);
+static_assert(std::ranges::view<std::ranges::take_view<MergeManyView<std::vector<std::vector<int>>>>>);
 
 template <std::ranges::input_range TRange>
 std::vector<TRange> ranges(TRange r1, TRange r2) {

silkworm/db/datastore/domain_range_as_of_query.hpp

@@ -38,27 +38,44 @@ struct DomainRangeAsOfQuery {
     using Key = decltype(TKeyEncoder1::value);
     using ResultItem = typename DomainRangeLatestQuery<TKeyEncoder1, TKeyEncoder2, TKeyDecoder1, TKeyDecoder2, TValueDecoder1, TValueDecoder2>::ResultItem;
 
-    auto exec(const Key& key_start, const Key& key_end, std::optional<Timestamp> timestamp, bool ascending) {
+    auto exec(const Key& key_start, const Key& key_end, std::optional<Timestamp> timestamp, bool ascending, bool skip_empty_values) {
         return silkworm::views::if_view(
             !timestamp.has_value(),
             query2_.exec(key_start, key_end, ascending),
-            this->exec(key_start, key_end, timestamp.value_or(0), ascending));
+            this->exec(key_start, key_end, timestamp.value_or(0), ascending, skip_empty_values));
     }
 
-    auto exec(const Key& key_start, const Key& key_end, Timestamp timestamp, bool ascending) {
+    auto exec(const Key& key_start, const Key& key_end, Timestamp timestamp, bool ascending, bool skip_empty_values) {
         SILKWORM_ASSERT(ascending);  // descending is not implemented
 
+        auto skip_empty_value_predicate = [skip_empty_values](std::pair<Bytes, Bytes>& kv_pair) {
+            if (!skip_empty_values) return true;
+            return !kv_pair.second.empty();
+        };
+
         return silkworm::views::merge_unique(
-            query1_.exec(key_start, key_end, timestamp, ascending),
-            query2_.exec(key_start, key_end, ascending),
-            silkworm::views::MergeCompareFunc{},
-            PairGetFirst<typename ResultItem::first_type, typename ResultItem::second_type>{},
-            PairGetFirst<typename ResultItem::first_type, typename ResultItem::second_type>{});
+                   query1_.exec(key_start, key_end, timestamp, ascending),
+                   query2_.exec(key_start, key_end, ascending),
+                   silkworm::views::MergeCompareFunc{},
+                   PairGetFirst<typename ResultItem::first_type, typename ResultItem::second_type>{},
+                   PairGetFirst<typename ResultItem::first_type, typename ResultItem::second_type>{}) |
+               std::views::filter(std::move(skip_empty_value_predicate));
     }
 
   private:
     HistoryRangeByKeysQuery<TKeyEncoder1, TKeyEncoder2, TKeyDecoder1, TKeyDecoder2, TValueDecoder1, TValueDecoder2> query1_;
     DomainRangeLatestQuery<TKeyEncoder1, TKeyEncoder2, TKeyDecoder1, TKeyDecoder2, TValueDecoder1, TValueDecoder2> query2_;
 };
 
+template <
+    kvdb::EncoderConcept TKeyEncoder1, snapshots::EncoderConcept TKeyEncoder2,
+    kvdb::DecoderConcept TKeyDecoder1, snapshots::DecoderConcept TKeyDecoder2,
+    kvdb::DecoderConcept TValueDecoder1, snapshots::DecoderConcept TValueDecoder2>
+using DomainRangeAsOfQueryResult = decltype(std::declval<DomainRangeAsOfQuery<TKeyEncoder1, TKeyEncoder2, TKeyDecoder1, TKeyDecoder2, TValueDecoder1, TValueDecoder2>>().exec(
+    std::declval<const typename DomainRangeAsOfQuery<TKeyEncoder1, TKeyEncoder2, TKeyDecoder1, TKeyDecoder2, TValueDecoder1, TValueDecoder2>::Key&>(),
+    std::declval<const typename DomainRangeAsOfQuery<TKeyEncoder1, TKeyEncoder2, TKeyDecoder1, TKeyDecoder2, TValueDecoder1, TValueDecoder2>::Key&>(),
+    std::declval<std::optional<Timestamp>>(),
+    std::declval<bool>(),
+    std::declval<bool>()));
+
 }  // namespace silkworm::datastore

silkworm/db/datastore/history_range_in_period_query.hpp

@@ -58,4 +58,11 @@ struct HistoryRangeInPeriodQuery {
     snapshots::HistoryRangeInPeriodQuery<TKeyDecoder2, TValueDecoder2> query2_;
 };
 
+template <
+    kvdb::DecoderConcept TKeyDecoder1, snapshots::DecoderConcept TKeyDecoder2,
+    kvdb::DecoderConcept TValueDecoder1, snapshots::DecoderConcept TValueDecoder2>
+using HistoryRangeInPeriodQueryResult = decltype(std::declval<HistoryRangeInPeriodQuery<TKeyDecoder1, TKeyDecoder2, TValueDecoder1, TValueDecoder2>>().exec(
+    std::declval<TimestampRange>(),
+    std::declval<bool>()));
+
 }  // namespace silkworm::datastore

silkworm/db/datastore/inverted_index_range_by_key_query.hpp

@@ -54,4 +54,10 @@ struct InvertedIndexRangeByKeyQuery {
     snapshots::InvertedIndexRangeByKeyQuery<TKeyEncoder2> query2_;
 };
 
+template <kvdb::EncoderConcept TKeyEncoder1, snapshots::EncoderConcept TKeyEncoder2>
+using InvertedIndexRangeByKeyQueryResult = decltype(std::declval<InvertedIndexRangeByKeyQuery<TKeyEncoder1, TKeyEncoder2>>().exec(
+    std::declval<typename InvertedIndexRangeByKeyQuery<TKeyEncoder1, TKeyEncoder2>::Key>(),
+    std::declval<TimestampRange>(),
+    std::declval<bool>()));
+
 }  // namespace silkworm::datastore

silkworm/db/datastore/kvdb/domain_range_latest_query.hpp

@@ -71,7 +71,7 @@ struct DomainRangeLatestQuery {
         }
 
         auto before_key_end_predicate = [key_end = std::move(key_end)](const std::pair<ByteView, ByteView>& kv_pair) {
-            return kv_pair.first < key_end;
+            return key_end.empty() || kv_pair.first < key_end;
         };
 
         return std::ranges::subrange{std::move(begin_it), CursorKVIteratorRaw{}} |

silkworm/db/datastore/kvdb/history_range_by_keys_query.hpp

@@ -126,6 +126,7 @@ struct HistoryRangeByKeysQuery {
         }
 
         auto before_key_end_predicate = [key_end = std::move(key_end)](const std::shared_ptr<ROCursor>& cursor) {
+            if (key_end.empty()) return true;
             auto result = cursor->current();
             SILKWORM_ASSERT(result);
             return from_slice(result.key) < ByteView{key_end};

silkworm/db/datastore/snapshots/domain_range_latest_query.hpp

@@ -39,7 +39,7 @@ struct DomainRangeLatestSegmentQuery {
         auto begin_it = entity_.btree_index.seek(key_start, entity_.kv_segment).value_or(btree::BTreeIndex::Cursor{});
 
         return std::ranges::subrange{std::move(begin_it), std::default_sentinel} |
-               std::views::take_while([key_end = std::move(key_end)](const auto& kv_pair) { return ByteView{kv_pair.first} < key_end; });
+               std::views::take_while([key_end = std::move(key_end)](const auto& kv_pair) { return key_end.empty() || ByteView{kv_pair.first} < key_end; });
     }
 
     auto exec(Bytes key_start, Bytes key_end, bool ascending) {
@@ -112,6 +112,7 @@ struct DomainRangeLatestQuery {
             PairGetFirst<DomainRangeLatestSegmentQuery::ResultItem::first_type, DomainRangeLatestSegmentQuery::ResultItem::second_type>{});
 
         return silkworm::ranges::owning_view(std::move(results)) |
+               std::views::filter([](const auto& kv_pair) { return !ByteView{kv_pair.second}.empty(); }) |
                std::views::transform(kDecodeKVPairFunc) |
                silkworm::views::caching;
     }

silkworm/db/datastore/snapshots/history_range_by_keys_query.hpp

@@ -71,7 +71,7 @@ struct HistoryRangeByKeysSegmentQuery {
         };
 
         return std::ranges::subrange{std::move(begin_it), ii_reader.end()} |
-               std::views::take_while([key_end = std::move(key_end)](auto&& ii_entry) { return ii_entry.first < key_end; }) |
+               std::views::take_while([key_end = std::move(key_end)](auto&& ii_entry) { return key_end.empty() || ii_entry.first < key_end; }) |
                std::views::transform(std::move(lookup_kv_pair_func)) |
                silkworm::views::caching |
                std::views::filter([](const std::optional<ResultItem>& result_opt) { return result_opt.has_value(); }) |