Implement asynchronous iterators for reading data in the storage. #4975

MathieuDutSik · 2025-11-16T12:48:27Z

Motivation

Accessing data at once can be suboptimal when reading data.
The solution to that is to implement asynchronous iterators.

Fixes #4831
Fixes #2742

Proposal

The implementation takes the keys by value (that is Vec<Vec<u8>>).
We cannot use references because when we create specific keys in
lru-caching and value-splitting, we would need to have a self-reference
which is not allowed.

The implementation by container is then as follows:

For DynamoDB and ScyllaDB, the downloading of data is done by block of 99 or 40 keys. So, the implementation of the iterator is straightforward to do.
For RocksDB / LineraStorageService, we download by blocks of 50 keys.This is somewhat artificial. But better than downloading all the data.
For lru-caching, value-splitting, we predetermine the needed keys. But when needed, sometimes we need to access keys (keys that were removed from the cache or additional segments) and so we access to the storage.

The implementation in the views is done for two cases where we have a use case in the code (the two issues in question):

Access to multiple keys in MapView.
Access to multiple views in try_load_all_entries.

For those two use cases, the loop can end prematurely, justifying the use of iterators.
It is unsure whether we can really reduce memory usage

Test Plan

The CI.

Release Plan

Those changes could be backported to TestNet Conway.

Links

None.

afck · 2025-11-17T09:54:35Z

linera-views/src/store.rs

+    /// Returns `Ok(None)` when iteration is complete.
+    /// Returns `Ok(Some(None))` when a key doesn't exist.
+    /// Returns `Ok(Some(Some(value)))` when a key exists with a value.
+    async fn next(&mut self) -> Result<Option<Option<Vec<u8>>>, E>;


Why not use Stream?

I.e. should ReadMultiIterator<E: KeyValueStoreError> just have no methods, and instead have impls and bounds so that it is equivalent to Stream<Item = Result<Option<Vec<u8>>, E> + MaybeSend where MaybeSend is Send iff not(web)?

If you write -> impl Stream<…> then trait_variant::make will do the MaybeSend part for you.

deuszx · 2025-11-17T11:59:39Z

linera-views/src/store.rs

+    /// Returns `Ok(None)` when iteration is complete.
+    /// Returns `Ok(Some(None))` when a key doesn't exist.


Why not the reverse? i.e. when key doesn't exist:

return Ok(None) when key doesn't exist. This is clearer b/c key doesn't exist form the first element. while Ok(Some(None)) could be returned for the subsequent calls to the iterator.

Ok(Some(None)) when there are no more elements in the stream.

No.
The function fn next returns a Option<Item> with Item being what is being produced. Here what is being produced is Option<Vec<u8>>.

if so, then why this signature isn't Option<Item> ? Where Item is a type member to be defined by the implementation.

Here Item = Option<Vec<u8>> which gets you Option<Option<Vec<u8>>>.

I understand. I mean that from Option<Option<Vec<u8>> you can't tell why you chose the semantics described in the comment wheras with Option<Item> it is.

~~Actually, even with Option<Item> and type Item = Option<Vec<u8>> it still makes more sense for None to mean the key doesn't exist, rather than Some(None) that you're proposing.~~

linera-views/src/lru_prefix_cache.rs

deuszx · 2025-11-17T12:02:53Z

linera-views/src/lru_prefix_cache.rs

+        if result {
+            return true;
+        }
+        if self.has_exclusive_access {


Why for self.has_exclusive_access == false we claim that key is not present?

If !self.has_exclusive_access then that array is empty so there is no point in checking in it.

linera-views/tests/store_tests.rs

linera-views/src/views/key_value_store_view.rs

afck · 2025-11-24T10:20:46Z

linera-views/src/test_utils/mod.rs

+        let mut values_from_iter = Vec::new();
+        loop {
+            match iter.next().await {
+                None => break,
+                Some(Ok(value)) => values_from_iter.push(value),
+                Some(Err(e)) => panic!("Error reading from iterator: {:?}", e),
+            }
+        }


Maybe more concise:

Suggested change

let mut values_from_iter = Vec::new();

loop {

match iter.next().await {

None => break,

Some(Ok(value)) => values_from_iter.push(value),

Some(Err(e)) => panic!("Error reading from iterator: {:?}", e),

}

}

let values_from_iter = iter

.collect::<Result<Vec<_>>>()

.await

.expect("reading from iterator"));

Yes, we can replace by

let values_from_iter = store .read_multi_values_bytes_iter(keys) .try_collect::<Vec<_>>() .await .expect("Error reading from iterator");

afck · 2025-11-24T10:26:42Z

linera-views/src/backends/value_splitting.rs


+    fn read_multi_values_bytes_iter(
+        &self,
+        keys: Vec<Vec<u8>>,


I wonder if it would make sense (maybe not in this PR) to also take a stream as an input. (Or just a regular iterator? Needs discussion…) That way, e.g. we wouldn't even put together the big_key for the 10000th element if the caller cancels the stream after the 5000th.

We potentially could have those everywhere. But I want to have a limited change here.

afck · 2025-11-24T10:28:39Z

linera-views/src/backends/value_splitting.rs

+                let value_result = first_segments_stream.next().await
+                    .expect("keys and first_segments should have same length");
+
+                let value = match value_result {


We should use try_stream! instead of stream!. Then this line could probably be let value = value_result?;, or let value = value_result.map_err(ValueSplittingError::InnerStoreError)? or something like that.

Yes, we can do that.
But no, we do not need the try_stream.

afck · 2025-11-24T10:29:46Z

linera-views/src/backends/scylla_db.rs

+            .map(|chunk| chunk.to_vec())
+            .collect::<Vec<_>>();


Suggested change

.map(|chunk| chunk.to_vec())

.collect::<Vec<_>>();

.map(|chunk| chunk.to_vec());

We need to collect the entries in order to be able to iterate on them. Removing the collect makes the code not compile.

That's just a borrow checker issue. This compiles, because it makes the stream own keys, and it avoids the allocation:

- let batches = keys - .chunks(MAX_MULTI_KEYS) - .map(|chunk| chunk.to_vec()) - .collect::<Vec<_>>(); let store = self.clone(); async_stream::stream! { - for batch_keys in batches { + for chunk in keys.chunks(MAX_MULTI_KEYS) { let vals = { + let batch_keys = chunk.to_vec();

afck · 2025-11-24T10:31:44Z

linera-views/src/backends/rocks_db.rs

+                let end_position = std::cmp::min(current_position + BATCH_SIZE, keys.len());
+
+                // Extract the next batch of keys
+                let batch_keys = keys[current_position..end_position].to_vec();


Why not use chunks here, too, like in the ScyllaDB case?

Yes, we can use the terminolology chunk systematically.

I meant the chunks method, so that you don't have to keep track of the positions manually here. (Or maybe there's a reason for it?)

afck · 2025-11-24T10:36:02Z

linera-views/src/backends/lru_caching.rs

+                            value
+                        } else {
+                            // The key has been evicted. Should be rare.
+                            store.read_value_bytes(key).await?


(Ah, that works? Then I'm not sure I understand the difference between stream and try_stream. Maybe just that in the latter case you write yield a instead of yield Ok(a)?)

Yes, that works. I do not need try_stream.

MathieuDutSik added 4 commits November 16, 2025 12:39

Introduce the read_multi_values_bytes_iter.

07a4b41

Some update.

c9959d5

Some errors from CI addressed.

1b01338

Simplification of the lifetime requirements.

6a0cd77

MathieuDutSik marked this pull request as ready for review November 16, 2025 19:36

MathieuDutSik requested review from afck, deuszx and ma2bd November 17, 2025 06:13

afck reviewed Nov 17, 2025

View reviewed changes

deuszx reviewed Nov 17, 2025

View reviewed changes

linera-views/src/lru_prefix_cache.rs Outdated Show resolved Hide resolved

deuszx reviewed Nov 17, 2025

View reviewed changes

MathieuDutSik added 6 commits November 17, 2025 13:40

Simplification of the test code.

53dba86

Update by using the async-stream.

013ca0e

Some reformatting and corrections.

86501e5

Some simplification of the read_multi_values_bytes_iter.

31a3c3e

Use the ? which is actually allowed in async_streams.

c12272c

Correct the cargo.lock file.

debced7

afck reviewed Nov 24, 2025

View reviewed changes

		/// Returns `Ok(None)` when iteration is complete.
		/// Returns `Ok(Some(None))` when a key doesn't exist.

	.map(\|chunk\| chunk.to_vec())
	.collect::<Vec<_>>();
	.map(\|chunk\| chunk.to_vec());

Implement asynchronous iterators for reading data in the storage. #4975

Are you sure you want to change the base?

Implement asynchronous iterators for reading data in the storage. #4975

Uh oh!

Conversation

MathieuDutSik commented Nov 16, 2025

Motivation

Proposal

Test Plan

Release Plan

Links

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

deuszx Nov 17, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

deuszx Nov 17, 2025 •

edited

Loading