Use query_iter

.github/workflows/scylla.yml

@@ -34,12 +34,10 @@ jobs:
           components: rustfmt, clippy
       - name: Load .env file
         uses: xom9ikk/dotenv@v1.0.2
+      - name: Run tests
+        # test threads must be one because else database tests will run in parallel and will result in flaky tests
+        run: cargo test --all -- --test-threads=1
       - name: Format check
         run: cargo fmt --all -- --check
       - name: Clippy check
-        run: cargo clippy --examples -- -D warnings
-      - name: Build
-        run: cargo build --examples
-      - name: Run tests
-        # test threads must be one because else database tests will run in parallel and will result in flaky tests
-        run: cargo test --all -- --test-threads=1
+        run: cargo clippy --examples -- -D warnings

scylla_orm/Cargo.toml

@@ -20,4 +20,5 @@ heck = "0.3"
 uuid = "0.8"
 tokio = { version = "1", features = ["time", "rt-multi-thread", "io-util"] }
 tracing = "0.1"
-thiserror = "1"
+thiserror = "1"
+futures-util = "0.3"

scylla_orm/src/query_transform.rs

@@ -1,5 +1,5 @@
-// Generated file
 use crate::Cursor;
+use futures_util::{StreamExt, TryStreamExt};
 use scylla::cql_to_rust::FromRowError;
 use scylla::frame::value::{SerializeValuesError, ValueList};
 use scylla::query::Query;
@@ -46,6 +46,11 @@ pub struct QueryEntityVecResult<T> {
     pub query_result: QueryResult,
 }
 
+/// Wrapper, maybe additional fields are added later
+pub struct QueryEntityVec<T> {
+    pub entities: Vec<T>,
+}
+
 impl<T> Deref for QueryEntityVecResult<T> {
     type Target = Vec<T>;
 
@@ -211,7 +216,7 @@ impl<R: AsRef<str>, V: ValueList> Qv<R, V> {
         session: &Session,
         page_size: i32,
         transform: impl Fn(T) -> N + Copy,
-    ) -> Result<QueryEntityVecResult<N>, MultipleSelectQueryErrorTransform> {
+    ) -> Result<QueryEntityVec<N>, MultipleSelectQueryErrorTransform> {
         let as_ref = self.query.as_ref();
 
         tracing::debug!("Executing with page size: {}: {}", page_size, as_ref);
@@ -220,13 +225,25 @@ impl<R: AsRef<str>, V: ValueList> Qv<R, V> {
 
         query.set_page_size(page_size);
 
-        let mut result = session.execute_all_cached(query, &self.values).await?;
-        let rows = self.transform(&mut result, transform)?;
+        let rows = session
+            .execute_iter_cached(query, &self.values)
+            .await?
+            .map(|c| {
+                let row = c.map(T::from_row);
+                let transformed = row.map(|r| r.map(|c| transform(c)));
+
+                match transformed {
+                    Ok(ok) => match ok {
+                        Ok(row) => Ok(row),
+                        Err(err) => Err(MultipleSelectQueryErrorTransform::FromRowError(err)),
+                    },
+                    Err(err) => Err(MultipleSelectQueryErrorTransform::QueryError(err)),
+                }
+            })
+            .try_collect::<Vec<_>>()
+            .await?;
 
-        Ok(QueryEntityVecResult {
-            entities: rows,
-            query_result: result,
-        })
+        Ok(QueryEntityVec { entities: rows })
     }
 
     async fn execute_iter_cached<T: FromRow>(
@@ -445,7 +462,7 @@ impl<R: AsRef<str>, T: FromRow, V: ValueList> SelectMultiple<R, T, V> {
         &self,
         session: &Session,
         page_size: i32,
-    ) -> Result<QueryEntityVecResult<T>, MultipleSelectQueryErrorTransform> {
+    ) -> Result<QueryEntityVec<T>, MultipleSelectQueryErrorTransform> {
         self.select_all_in_memory_transform(session, page_size, |v| v)
             .await
     }
@@ -455,7 +472,7 @@ impl<R: AsRef<str>, T: FromRow, V: ValueList> SelectMultiple<R, T, V> {
         session: &Session,
         page_size: i32,
         transform: impl Fn(T) -> N + Copy,
-    ) -> Result<QueryEntityVecResult<N>, MultipleSelectQueryErrorTransform> {
+    ) -> Result<QueryEntityVec<N>, MultipleSelectQueryErrorTransform> {
         self.qv
             .execute_all_in_memory_cached(session, page_size, transform)
             .await

scylla_orm_table_to_struct/example/src/generated/another_test_table.rs

@@ -7,9 +7,10 @@ use scylla::transport::iterator::TypedRowIterator;
 use scylla::Session;
 #[allow(unused_imports)]
 use scylla_orm::query_transform::{
-    CountType, DeleteUnique, Insert, MultipleSelectQueryErrorTransform, QueryEntityVecResult,
-    QueryResultUniqueRow, QueryResultUniqueRowExpect, Qv, ScyllaQueryResult, SelectMultiple,
-    SelectUnique, SelectUniqueExpect, SingleSelectQueryErrorTransform, Truncate, TtlType, Update,
+    CountType, DeleteUnique, Insert, MultipleSelectQueryErrorTransform, QueryEntityVec,
+    QueryEntityVecResult, QueryResultUniqueRow, QueryResultUniqueRowExpect, Qv, ScyllaQueryResult,
+    SelectMultiple, SelectUnique, SelectUniqueExpect, SingleSelectQueryErrorTransform, Truncate,
+    TtlType, Update,
 };
 pub const SELECT_ALL_QUERY: &str = "select a, b, c, d from another_test_table";
 pub const SELECT_ALL_COUNT_QUERY: &str = "select count(*) from another_test_table";
@@ -77,7 +78,7 @@ pub async fn select_all(
 pub async fn select_all_in_memory(
     session: &Session,
     page_size: i32,
-) -> Result<QueryEntityVecResult<AnotherTestTable>, MultipleSelectQueryErrorTransform> {
+) -> Result<QueryEntityVec<AnotherTestTable>, MultipleSelectQueryErrorTransform> {
     select_all_qv()
         .select_all_in_memory(session, page_size)
         .await

scylla_orm_table_to_struct/example/src/generated/child.rs

@@ -7,9 +7,10 @@ use scylla::transport::iterator::TypedRowIterator;
 use scylla::Session;
 #[allow(unused_imports)]
 use scylla_orm::query_transform::{
-    CountType, DeleteUnique, Insert, MultipleSelectQueryErrorTransform, QueryEntityVecResult,
-    QueryResultUniqueRow, QueryResultUniqueRowExpect, Qv, ScyllaQueryResult, SelectMultiple,
-    SelectUnique, SelectUniqueExpect, SingleSelectQueryErrorTransform, Truncate, TtlType, Update,
+    CountType, DeleteUnique, Insert, MultipleSelectQueryErrorTransform, QueryEntityVec,
+    QueryEntityVecResult, QueryResultUniqueRow, QueryResultUniqueRowExpect, Qv, ScyllaQueryResult,
+    SelectMultiple, SelectUnique, SelectUniqueExpect, SingleSelectQueryErrorTransform, Truncate,
+    TtlType, Update,
 };
 pub const SELECT_ALL_QUERY: &str = "select birthday, enum_json, json, json_nullable from child";
 pub const SELECT_ALL_COUNT_QUERY: &str = "select count(*) from child";
@@ -77,7 +78,7 @@ pub async fn select_all(
 pub async fn select_all_in_memory(
     session: &Session,
     page_size: i32,
-) -> Result<QueryEntityVecResult<Child>, MultipleSelectQueryErrorTransform> {
+) -> Result<QueryEntityVec<Child>, MultipleSelectQueryErrorTransform> {
     select_all_qv()
         .select_all_in_memory(session, page_size)
         .await

scylla_orm_table_to_struct/example/src/generated/person.rs

@@ -7,9 +7,10 @@ use scylla::transport::iterator::TypedRowIterator;
 use scylla::Session;
 #[allow(unused_imports)]
 use scylla_orm::query_transform::{
-    CountType, DeleteUnique, Insert, MultipleSelectQueryErrorTransform, QueryEntityVecResult,
-    QueryResultUniqueRow, QueryResultUniqueRowExpect, Qv, ScyllaQueryResult, SelectMultiple,
-    SelectUnique, SelectUniqueExpect, SingleSelectQueryErrorTransform, Truncate, TtlType, Update,
+    CountType, DeleteUnique, Insert, MultipleSelectQueryErrorTransform, QueryEntityVec,
+    QueryEntityVecResult, QueryResultUniqueRow, QueryResultUniqueRowExpect, Qv, ScyllaQueryResult,
+    SelectMultiple, SelectUnique, SelectUniqueExpect, SingleSelectQueryErrorTransform, Truncate,
+    TtlType, Update,
 };
 pub const SELECT_ALL_QUERY: &str = "select name, age, email from person";
 pub const SELECT_ALL_COUNT_QUERY: &str = "select count(*) from person";
@@ -72,7 +73,7 @@ pub async fn select_all(
 pub async fn select_all_in_memory(
     session: &Session,
     page_size: i32,
-) -> Result<QueryEntityVecResult<Person>, MultipleSelectQueryErrorTransform> {
+) -> Result<QueryEntityVec<Person>, MultipleSelectQueryErrorTransform> {
     select_all_qv()
         .select_all_in_memory(session, page_size)
         .await

scylla_orm_table_to_struct/example/src/generated/person_by_email.rs

@@ -8,9 +8,10 @@ use scylla::transport::iterator::TypedRowIterator;
 use scylla::Session;
 #[allow(unused_imports)]
 use scylla_orm::query_transform::{
-    CountType, DeleteUnique, Insert, MultipleSelectQueryErrorTransform, QueryEntityVecResult,
-    QueryResultUniqueRow, QueryResultUniqueRowExpect, Qv, ScyllaQueryResult, SelectMultiple,
-    SelectUnique, SelectUniqueExpect, SingleSelectQueryErrorTransform, Truncate, TtlType, Update,
+    CountType, DeleteUnique, Insert, MultipleSelectQueryErrorTransform, QueryEntityVec,
+    QueryEntityVecResult, QueryResultUniqueRow, QueryResultUniqueRowExpect, Qv, ScyllaQueryResult,
+    SelectMultiple, SelectUnique, SelectUniqueExpect, SingleSelectQueryErrorTransform, Truncate,
+    TtlType, Update,
 };
 pub const SELECT_ALL_QUERY: &str = "select email, name, age from person_by_email";
 pub const SELECT_ALL_COUNT_QUERY: &str = "select count(*) from person_by_email";
@@ -73,7 +74,7 @@ pub async fn select_all(
 pub async fn select_all_in_memory(
     session: &Session,
     page_size: i32,
-) -> Result<QueryEntityVecResult<PersonByEmail>, MultipleSelectQueryErrorTransform> {
+) -> Result<QueryEntityVec<PersonByEmail>, MultipleSelectQueryErrorTransform> {
     select_all_qv()
         .select_all_in_memory(session, page_size)
         .await
@@ -126,7 +127,7 @@ pub async fn select_all_base_table(
 pub async fn select_all_base_table_in_memory(
     session: &Session,
     page_size: i32,
-) -> Result<QueryEntityVecResult<Person>, MultipleSelectQueryErrorTransform> {
+) -> Result<QueryEntityVec<Person>, MultipleSelectQueryErrorTransform> {
     select_all_base_table_qv()
         .select_all_in_memory(session, page_size)
         .await

scylla_orm_table_to_struct/example/src/generated/test_table.rs

@@ -7,9 +7,10 @@ use scylla::transport::iterator::TypedRowIterator;
 use scylla::Session;
 #[allow(unused_imports)]
 use scylla_orm::query_transform::{
-    CountType, DeleteUnique, Insert, MultipleSelectQueryErrorTransform, QueryEntityVecResult,
-    QueryResultUniqueRow, QueryResultUniqueRowExpect, Qv, ScyllaQueryResult, SelectMultiple,
-    SelectUnique, SelectUniqueExpect, SingleSelectQueryErrorTransform, Truncate, TtlType, Update,
+    CountType, DeleteUnique, Insert, MultipleSelectQueryErrorTransform, QueryEntityVec,
+    QueryEntityVecResult, QueryResultUniqueRow, QueryResultUniqueRowExpect, Qv, ScyllaQueryResult,
+    SelectMultiple, SelectUnique, SelectUniqueExpect, SingleSelectQueryErrorTransform, Truncate,
+    TtlType, Update,
 };
 pub const SELECT_ALL_QUERY: &str = "select b, c, d, a, e from test_table";
 pub const SELECT_ALL_COUNT_QUERY: &str = "select count(*) from test_table";
@@ -81,7 +82,7 @@ pub async fn select_all(
 pub async fn select_all_in_memory(
     session: &Session,
     page_size: i32,
-) -> Result<QueryEntityVecResult<TestTable>, MultipleSelectQueryErrorTransform> {
+) -> Result<QueryEntityVec<TestTable>, MultipleSelectQueryErrorTransform> {
     select_all_qv()
         .select_all_in_memory(session, page_size)
         .await

scylla_orm_table_to_struct/example/src/generated/uuidtable.rs

@@ -7,9 +7,10 @@ use scylla::transport::iterator::TypedRowIterator;
 use scylla::Session;
 #[allow(unused_imports)]
 use scylla_orm::query_transform::{
-    CountType, DeleteUnique, Insert, MultipleSelectQueryErrorTransform, QueryEntityVecResult,
-    QueryResultUniqueRow, QueryResultUniqueRowExpect, Qv, ScyllaQueryResult, SelectMultiple,
-    SelectUnique, SelectUniqueExpect, SingleSelectQueryErrorTransform, Truncate, TtlType, Update,
+    CountType, DeleteUnique, Insert, MultipleSelectQueryErrorTransform, QueryEntityVec,
+    QueryEntityVecResult, QueryResultUniqueRow, QueryResultUniqueRowExpect, Qv, ScyllaQueryResult,
+    SelectMultiple, SelectUnique, SelectUniqueExpect, SingleSelectQueryErrorTransform, Truncate,
+    TtlType, Update,
 };
 pub const SELECT_ALL_QUERY: &str = "select u from uuidtable";
 pub const SELECT_ALL_COUNT_QUERY: &str = "select count(*) from uuidtable";
@@ -60,7 +61,7 @@ pub async fn select_all(
 pub async fn select_all_in_memory(
     session: &Session,
     page_size: i32,
-) -> Result<QueryEntityVecResult<Uuidtable>, MultipleSelectQueryErrorTransform> {
+) -> Result<QueryEntityVec<Uuidtable>, MultipleSelectQueryErrorTransform> {
     select_all_qv()
         .select_all_in_memory(session, page_size)
         .await

scylla_orm_table_to_struct/src/entity_writer.rs

@@ -43,6 +43,7 @@ impl<T: Transformer> EntityWriter<'_, T> {
                 ScyllaQueryResult,
                 SingleSelectQueryErrorTransform,
                 QueryEntityVecResult,
+                QueryEntityVec,
                 MultipleSelectQueryErrorTransform,
                 QueryResultUniqueRow,
                 QueryResultUniqueRowExpect,

scylla_orm_table_to_struct/src/entity_writer/write_struct.rs

@@ -301,7 +301,7 @@ fn create_select_all_query(
             #select_multiple_qv().select(session, page_size).await
         }
 
-        pub async fn #select_multiple_all_in_memory(session: &Session, page_size: i32) -> Result<QueryEntityVecResult<#row_type>, MultipleSelectQueryErrorTransform> {
+        pub async fn #select_multiple_all_in_memory(session: &Session, page_size: i32) -> Result<QueryEntityVec<#row_type>, MultipleSelectQueryErrorTransform> {
             #select_multiple_qv().select_all_in_memory(session, page_size).await
         }
     }