Merge pull request #478 from gwenn/fallible-iterator

Make `Rows` implement `FallibleStreamingIterator`

Cargo.toml

@@ -47,7 +47,7 @@ csvtab = ["csv", "vtab"]
 # pointer passing interfaces: 3.20.0
 array = ["vtab"]
 # session extension: 3.13.0
-session = ["libsqlite3-sys/session", "hooks", "fallible-streaming-iterator"]
+session = ["libsqlite3-sys/session", "hooks"]
 
 [dependencies]
 time = "0.1.0"
@@ -59,7 +59,8 @@ csv = { version = "1.0", optional = true }
 url = { version = "1.7", optional = true }
 lazy_static = { version = "1.0", optional = true }
 byteorder = { version = "1.2", features = ["i128"], optional = true }
-fallible-streaming-iterator = { version = "0.1", optional = true }
+fallible-iterator = "0.1"
+fallible-streaming-iterator = "0.1"
 memchr = "2.2.0"
 
 [dev-dependencies]

src/cache.rs

@@ -151,6 +151,7 @@ impl StatementCache {
 
 #[cfg(test)]
 mod test {
+    use fallible_iterator::FallibleIterator;
     use super::StatementCache;
     use crate::{Connection, NO_PARAMS};
 
@@ -277,12 +278,9 @@ mod test {
         {
             let mut stmt = db.prepare_cached(sql).unwrap();
             assert_eq!(
-                1i32,
+                Ok(Some(1i32)),
-                stmt.query_map::<i32, _, _>(NO_PARAMS, |r| r.get(0))
+                stmt.query(NO_PARAMS).unwrap().map(|r| r.get(0))
-                    .unwrap()
                     .next()
-                    .unwrap()
-                    .unwrap()
             );
         }
 
@@ -297,12 +295,9 @@ mod test {
         {
             let mut stmt = db.prepare_cached(sql).unwrap();
             assert_eq!(
-                (1i32, 2i32),
+                Ok(Some((1i32, 2i32))),
-                stmt.query_map(NO_PARAMS, |r| Ok((r.get(0)?, r.get(1)?)))
+                stmt.query(NO_PARAMS).unwrap().map(|r| Ok((r.get(0)?, r.get(1)?)))
-                    .unwrap()
                     .next()
-                    .unwrap()
-                    .unwrap()
             );
         }
     }

src/inner_connection.rs

@@ -9,7 +9,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, Once, ONCE_INIT};
 
 use super::ffi;
-use super::{str_to_cstring, str_for_sqlite};
+use super::{str_for_sqlite, str_to_cstring};
 use super::{Connection, InterruptHandle, OpenFlags, Result};
 use crate::error::{error_from_handle, error_from_sqlite_code, Error};
 use crate::raw_statement::RawStatement;
@@ -230,13 +230,7 @@ impl InnerConnection {
                 }
                 rc
             } else {
-                ffi::sqlite3_prepare_v2(
+                ffi::sqlite3_prepare_v2(self.db(), c_sql, len, &mut c_stmt, ptr::null_mut())
-                    self.db(),
-                    c_sql,
-                    len,
-                    &mut c_stmt,
-                    ptr::null_mut(),
-                )
             }
         };
         self.decode_result(r)

src/lib.rs

@@ -86,7 +86,7 @@ pub use crate::cache::CachedStatement;
 pub use crate::error::Error;
 pub use crate::ffi::ErrorCode;
 #[cfg(feature = "hooks")]
-pub use crate::hooks::*;
+pub use crate::hooks::Action;
 #[cfg(feature = "load_extension")]
 pub use crate::load_extension_guard::LoadExtensionGuard;
 pub use crate::row::{AndThenRows, MappedRows, Row, RowIndex, Rows};
@@ -265,7 +265,10 @@ fn str_for_sqlite(s: &str) -> Result<(*const c_char, c_int, ffi::sqlite3_destruc
 // failed.
 fn len_as_c_int(len: usize) -> Result<c_int> {
     if len >= (c_int::max_value() as usize) {
-        Err(Error::SqliteFailure(ffi::Error::new(ffi::SQLITE_TOOBIG), None))
+        Err(Error::SqliteFailure(
+            ffi::Error::new(ffi::SQLITE_TOOBIG),
+            None,
+        ))
     } else {
         Ok(len as c_int)
     }
@@ -507,7 +510,7 @@ impl Connection {
     where
         P: IntoIterator,
         P::Item: ToSql,
-        F: FnOnce(&Row<'_, '_>) -> Result<T>,
+        F: FnOnce(&Row<'_>) -> Result<T>,
     {
         let mut stmt = self.prepare(sql)?;
         stmt.query_row(params, f)
@@ -529,7 +532,7 @@ impl Connection {
     /// or if the underlying SQLite call fails.
     pub fn query_row_named<T, F>(&self, sql: &str, params: &[(&str, &dyn ToSql)], f: F) -> Result<T>
     where
-        F: FnOnce(&Row<'_, '_>) -> Result<T>,
+        F: FnOnce(&Row<'_>) -> Result<T>,
     {
         let mut stmt = self.prepare(sql)?;
         let mut rows = stmt.query_named(params)?;
@@ -566,7 +569,7 @@ impl Connection {
     where
         P: IntoIterator,
         P::Item: ToSql,
-        F: FnOnce(&Row<'_, '_>) -> result::Result<T, E>,
+        F: FnOnce(&Row<'_>) -> result::Result<T, E>,
         E: convert::From<Error>,
     {
         let mut stmt = self.prepare(sql)?;
@@ -841,6 +844,7 @@ unsafe fn db_filename(_: *mut ffi::sqlite3) -> Option<PathBuf> {
 
 #[cfg(test)]
 mod test {
+    use fallible_iterator::FallibleIterator;
     use self::tempdir::TempDir;
     pub use super::*;
     use crate::ffi;
@@ -1094,8 +1098,8 @@ mod test {
             let mut rows = query.query(&[4i32]).unwrap();
             let mut v = Vec::<i32>::new();
 
-            while let Some(row) = rows.next() {
+            while let Some(row) = rows.next().unwrap() {
-                v.push(row.unwrap().get(0).unwrap());
+                v.push(row.get(0).unwrap());
             }
 
             assert_eq!(v, [3i32, 2, 1]);
@@ -1105,8 +1109,8 @@ mod test {
             let mut rows = query.query(&[3i32]).unwrap();
             let mut v = Vec::<i32>::new();
 
-            while let Some(row) = rows.next() {
+            while let Some(row) = rows.next().unwrap() {
-                v.push(row.unwrap().get(0).unwrap());
+                v.push(row.get(0).unwrap());
             }
 
             assert_eq!(v, [2i32, 1]);
@@ -1127,8 +1131,7 @@ mod test {
 
         let mut query = db.prepare("SELECT x, y FROM foo ORDER BY x DESC").unwrap();
         let results: Result<Vec<String>> = query
-            .query_map(NO_PARAMS, |row| row.get(1))
+            .query(NO_PARAMS).unwrap().map(|row| row.get(1))
-            .unwrap()
             .collect();
 
         assert_eq!(results.unwrap().concat(), "hello, world!");
@@ -1248,7 +1251,7 @@ mod test {
         {
             let mut rows = stmt.query(NO_PARAMS).unwrap();
             assert!(!db.is_busy());
-            let row = rows.next();
+            let row = rows.next().unwrap();
             assert!(db.is_busy());
             assert!(row.is_some());
         }
@@ -1317,7 +1320,7 @@ mod test {
             .prepare("SELECT interrupt() FROM (SELECT 1 UNION SELECT 2 UNION SELECT 3)")
             .unwrap();
 
-        let result: Result<Vec<i32>> = stmt.query_map(NO_PARAMS, |r| r.get(0)).unwrap().collect();
+        let result: Result<Vec<i32>> = stmt.query(NO_PARAMS).unwrap().map(|r| r.get(0)).collect();
 
         match result.unwrap_err() {
             Error::SqliteFailure(err, _) => {
@@ -1360,8 +1363,7 @@ mod test {
         let mut query = db.prepare("SELECT i, x FROM foo").unwrap();
         let mut rows = query.query(NO_PARAMS).unwrap();
 
-        while let Some(res) = rows.next() {
+        while let Some(row) = rows.next().unwrap() {
-            let row = res.unwrap();
             let i = row.get_raw(0).as_i64().unwrap();
             let expect = vals[i as usize];
             let x = row.get_raw("x").as_str().unwrap();

src/pragma.rs

@@ -171,7 +171,7 @@ impl Connection {
         f: F,
     ) -> Result<T>
     where
-        F: FnOnce(&Row<'_, '_>) -> Result<T>,
+        F: FnOnce(&Row<'_>) -> Result<T>,
     {
         let mut query = Sql::new();
         query.push_pragma(schema_name, pragma_name)?;
@@ -189,14 +189,14 @@ impl Connection {
         mut f: F,
     ) -> Result<()>
     where
-        F: FnMut(&Row<'_, '_>) -> Result<()>,
+        F: FnMut(&Row<'_>) -> Result<()>,
     {
         let mut query = Sql::new();
         query.push_pragma(schema_name, pragma_name)?;
         let mut stmt = self.prepare(&query)?;
         let mut rows = stmt.query(NO_PARAMS)?;
-        while let Some(result_row) = rows.next() {
+        while let Some(result_row) = rows.next()? {
-            let row = result_row?;
+            let row = result_row;
             f(&row)?;
         }
         Ok(())
@@ -219,7 +219,7 @@ impl Connection {
         mut f: F,
     ) -> Result<()>
     where
-        F: FnMut(&Row<'_, '_>) -> Result<()>,
+        F: FnMut(&Row<'_>) -> Result<()>,
     {
         let mut sql = Sql::new();
         sql.push_pragma(schema_name, pragma_name)?;
@@ -231,8 +231,8 @@ impl Connection {
         sql.close_brace();
         let mut stmt = self.prepare(&sql)?;
         let mut rows = stmt.query(NO_PARAMS)?;
-        while let Some(result_row) = rows.next() {
+        while let Some(result_row) = rows.next()? {
-            let row = result_row?;
+            let row = result_row;
             f(&row)?;
         }
         Ok(())
@@ -269,7 +269,7 @@ impl Connection {
         f: F,
     ) -> Result<T>
     where
-        F: FnOnce(&Row<'_, '_>) -> Result<T>,
+        F: FnOnce(&Row<'_>) -> Result<T>,
     {
         let mut sql = Sql::new();
         sql.push_pragma(schema_name, pragma_name)?;
@@ -385,8 +385,8 @@ mod test {
         let mut columns = Vec::new();
         let mut rows = table_info.query(&["sqlite_master"]).unwrap();
 
-        while let Some(row) = rows.next() {
+        while let Some(row) = rows.next().unwrap() {
-            let row = row.unwrap();
+            let row = row;
             let column: String = row.get(1).unwrap();
             columns.push(column);
         }

src/row.rs

@@ -1,4 +1,5 @@
-use std::marker::PhantomData;
+use fallible_iterator::FallibleIterator;
+use fallible_streaming_iterator::FallibleStreamingIterator;
 use std::{convert, result};
 
 use super::{Error, Result, Statement};
@@ -7,6 +8,7 @@ use crate::types::{FromSql, FromSqlError, ValueRef};
 /// An handle for the resulting rows of a query.
 pub struct Rows<'stmt> {
     stmt: Option<&'stmt Statement<'stmt>>,
+    row: Option<Row<'stmt>>,
 }
 
 impl<'stmt> Rows<'stmt> {
@@ -16,44 +18,42 @@ impl<'stmt> Rows<'stmt> {
         }
     }
 
-    /// Attempt to get the next row from the query. Returns `Some(Ok(Row))` if
+    /// Attempt to get the next row from the query. Returns `Ok(Some(Row))` if
-    /// there is another row, `Some(Err(...))` if there was an error
+    /// there is another row, `Err(...)` if there was an error
-    /// getting the next row, and `None` if all rows have been retrieved.
+    /// getting the next row, and `Ok(None)` if all rows have been retrieved.
     ///
     /// ## Note
     ///
     /// This interface is not compatible with Rust's `Iterator` trait, because
     /// the lifetime of the returned row is tied to the lifetime of `self`.
-    /// This is a "streaming iterator". For a more natural interface,
+    /// This is a fallible "streaming iterator". For a more natural interface,
     /// consider using `query_map` or `query_and_then` instead, which
     /// return types that implement `Iterator`.
     #[allow(clippy::should_implement_trait)] // cannot implement Iterator
-    pub fn next<'a>(&'a mut self) -> Option<Result<Row<'a, 'stmt>>> {
+    pub fn next(&mut self) -> Result<Option<&Row<'stmt>>> {
-        self.stmt.and_then(|stmt| match stmt.step() {
+        self.advance()?;
-            Ok(true) => Some(Ok(Row {
+        Ok((*self).get())
-                stmt,
-                phantom: PhantomData,
-            })),
-            Ok(false) => {
-                self.reset();
-                None
     }
-            Err(err) => {
+
-                self.reset();
+    pub fn map<F, B>(self, f: F) -> Map<'stmt, F>
-                Some(Err(err))
+    where
-            }
+        F: FnMut(&Row<'_>) -> Result<B>,
-        })
+    {
+        Map { rows: self, f: f }
     }
 }
 
 impl<'stmt> Rows<'stmt> {
     pub(crate) fn new(stmt: &'stmt Statement<'stmt>) -> Rows<'stmt> {
-        Rows { stmt: Some(stmt) }
+        Rows {
+            stmt: Some(stmt),
+            row: None,
+        }
     }
 
-    pub(crate) fn get_expected_row<'a>(&'a mut self) -> Result<Row<'a, 'stmt>> {
+    pub(crate) fn get_expected_row(&mut self) -> Result<&Row<'stmt>> {
-        match self.next() {
+        match self.next()? {
-            Some(row) => row,
+            Some(row) => Ok(row),
             None => Err(Error::QueryReturnedNoRows),
         }
     }
@@ -65,6 +65,26 @@ impl Drop for Rows<'_> {
     }
 }
 
+pub struct Map<'stmt, F> {
+    rows: Rows<'stmt>,
+    f: F,
+}
+
+impl<F, B> FallibleIterator for Map<'_, F>
+where
+    F: FnMut(&Row<'_>) -> Result<B>,
+{
+    type Error = Error;
+    type Item = B;
+
+    fn next(&mut self) -> Result<Option<B>> {
+        match self.rows.next()? {
+            Some(v) => Ok(Some((self.f)(v)?)),
+            None => Ok(None),
+        }
+    }
+}
+
 /// An iterator over the mapped resulting rows of a query.
 pub struct MappedRows<'stmt, F> {
     rows: Rows<'stmt>,
@@ -73,7 +93,7 @@ pub struct MappedRows<'stmt, F> {
 
 impl<'stmt, T, F> MappedRows<'stmt, F>
 where
-    F: FnMut(&Row<'_, '_>) -> Result<T>,
+    F: FnMut(&Row<'_>) -> Result<T>,
 {
     pub(crate) fn new(rows: Rows<'stmt>, f: F) -> MappedRows<'stmt, F> {
         MappedRows { rows, map: f }
@@ -82,7 +102,7 @@ where
 
 impl<T, F> Iterator for MappedRows<'_, F>
 where
-    F: FnMut(&Row<'_, '_>) -> Result<T>,
+    F: FnMut(&Row<'_>) -> Result<T>,
 {
     type Item = Result<T>;
 
@@ -90,6 +110,7 @@ where
         let map = &mut self.map;
         self.rows
             .next()
+            .transpose()
             .map(|row_result| row_result.and_then(|row| (map)(&row)))
     }
 }
@@ -103,7 +124,7 @@ pub struct AndThenRows<'stmt, F> {
 
 impl<'stmt, T, E, F> AndThenRows<'stmt, F>
 where
-    F: FnMut(&Row<'_, '_>) -> result::Result<T, E>,
+    F: FnMut(&Row<'_>) -> result::Result<T, E>,
 {
     pub(crate) fn new(rows: Rows<'stmt>, f: F) -> AndThenRows<'stmt, F> {
         AndThenRows { rows, map: f }
@@ -113,7 +134,7 @@ where
 impl<T, E, F> Iterator for AndThenRows<'_, F>
 where
     E: convert::From<Error>,
-    F: FnMut(&Row<'_, '_>) -> result::Result<T, E>,
+    F: FnMut(&Row<'_>) -> result::Result<T, E>,
 {
     type Item = result::Result<T, E>;
 
@@ -121,17 +142,51 @@ where
         let map = &mut self.map;
         self.rows
             .next()
+            .transpose()
             .map(|row_result| row_result.map_err(E::from).and_then(|row| (map)(&row)))
     }
 }
 
-/// A single result row of a query.
+impl<'stmt> FallibleStreamingIterator for Rows<'stmt> {
-pub struct Row<'a, 'stmt: 'a> {
+    type Error = Error;
-    stmt: &'stmt Statement<'stmt>,
+    type Item = Row<'stmt>;
-    phantom: PhantomData<&'a ()>,
+
+    fn advance(&mut self) -> Result<()> {
+        match self.stmt {
+            Some(ref stmt) => match stmt.step() {
+                Ok(true) => {
+                    self.row = Some(Row { stmt });
+                    Ok(())
+                }
+                Ok(false) => {
+                    self.reset();
+                    self.row = None;
+                    Ok(())
+                }
+                Err(e) => {
+                    self.reset();
+                    self.row = None;
+                    Err(e)
+                }
+            },
+            None => {
+                self.row = None;
+                Ok(())
+            }
+        }
+    }
+
+    fn get(&self) -> Option<&Row<'stmt>> {
+        self.row.as_ref()
+    }
 }
 
-impl<'a, 'stmt> Row<'a, 'stmt> {
+/// A single result row of a query.
+pub struct Row<'stmt> {
+    stmt: &'stmt Statement<'stmt>,
+}
+
+impl<'stmt> Row<'stmt> {
     /// Get the value of a particular column of the result row.
     ///
     /// ## Failure
@@ -193,7 +248,7 @@ impl<'a, 'stmt> Row<'a, 'stmt> {
     ///
     /// Returns an `Error::InvalidColumnName` if `idx` is not a valid column
     /// name for this row.
-    pub fn get_raw_checked<I: RowIndex>(&self, idx: I) -> Result<ValueRef<'a>> {
+    pub fn get_raw_checked<I: RowIndex>(&self, idx: I) -> Result<ValueRef<'_>> {
         let idx = idx.idx(self.stmt)?;
         // Narrowing from `ValueRef<'stmt>` (which `self.stmt.value_ref(idx)`
         // returns) to `ValueRef<'a>` is needed because it's only valid until
@@ -217,7 +272,7 @@ impl<'a, 'stmt> Row<'a, 'stmt> {
     ///
     /// * If `idx` is outside the range of columns in the returned query.
     /// * If `idx` is not a valid column name for this row.
-    pub fn get_raw<I: RowIndex>(&self, idx: I) -> ValueRef<'a> {
+    pub fn get_raw<I: RowIndex>(&self, idx: I) -> ValueRef<'_> {
         self.get_raw_checked(idx).unwrap()
     }
 

src/session.rs

@@ -719,8 +719,8 @@ unsafe extern "C" fn x_output(p_out: *mut c_void, data: *const c_void, len: c_in
 
 #[cfg(test)]
 mod test {
-    use fallible_streaming_iterator::FallibleStreamingIterator;
     use std::sync::atomic::{AtomicBool, Ordering};
+    use fallible_streaming_iterator::FallibleStreamingIterator;
 
     use super::{Changeset, ChangesetIter, ConflictAction, ConflictType, Session};
     use crate::hooks::Action;

src/statement.rs

@@ -7,7 +7,7 @@ use std::slice::from_raw_parts;
 use std::{convert, fmt, mem, ptr, result, str};
 
 use super::ffi;
-use super::{str_to_cstring, str_for_sqlite, len_as_c_int};
+use super::{len_as_c_int, str_for_sqlite, str_to_cstring};
 use super::{
     AndThenRows, Connection, Error, MappedRows, RawStatement, Result, Row, Rows, ValueRef,
 };
@@ -174,8 +174,7 @@ impl Statement<'_> {
     ///     let mut rows = stmt.query(NO_PARAMS)?;
     ///
     ///     let mut names = Vec::new();
-    ///     while let Some(result_row) = rows.next() {
+    ///     while let Some(row) = rows.next()? {
-    ///         let row = result_row?;
     ///         names.push(row.get(0)?);
     ///     }
     ///
@@ -209,7 +208,7 @@ impl Statement<'_> {
     /// fn query(conn: &Connection) -> Result<()> {
     ///     let mut stmt = conn.prepare("SELECT * FROM test where name = :name")?;
     ///     let mut rows = stmt.query_named(&[(":name", &"one")])?;
-    ///     while let Some(row) = rows.next() {
+    ///     while let Some(row) = rows.next()? {
     ///         // ...
     ///     }
     ///     Ok(())
@@ -224,7 +223,7 @@ impl Statement<'_> {
     /// fn query(conn: &Connection) -> Result<()> {
     ///     let mut stmt = conn.prepare("SELECT * FROM test where name = :name")?;
     ///     let mut rows = stmt.query_named(named_params!{ ":name": "one" })?;
-    ///     while let Some(row) = rows.next() {
+    ///     while let Some(row) = rows.next()? {
     ///         // ...
     ///     }
     ///     Ok(())
@@ -267,7 +266,7 @@ impl Statement<'_> {
     where
         P: IntoIterator,
         P::Item: ToSql,
-        F: FnMut(&Row<'_, '_>) -> Result<T>,
+        F: FnMut(&Row<'_>) -> Result<T>,
     {
         let rows = self.query(params)?;
         Ok(MappedRows::new(rows, f))
@@ -306,7 +305,7 @@ impl Statement<'_> {
         f: F,
     ) -> Result<MappedRows<'_, F>>
     where
-        F: FnMut(&Row<'_, '_>) -> Result<T>,
+        F: FnMut(&Row<'_>) -> Result<T>,
     {
         let rows = self.query_named(params)?;
         Ok(MappedRows::new(rows, f))
@@ -324,7 +323,7 @@ impl Statement<'_> {
         P: IntoIterator,
         P::Item: ToSql,
         E: convert::From<Error>,
-        F: FnMut(&Row<'_, '_>) -> result::Result<T, E>,
+        F: FnMut(&Row<'_>) -> result::Result<T, E>,
     {
         let rows = self.query(params)?;
         Ok(AndThenRows::new(rows, f))
@@ -375,7 +374,7 @@ impl Statement<'_> {
     ) -> Result<AndThenRows<'_, F>>
     where
         E: convert::From<Error>,
-        F: FnMut(&Row<'_, '_>) -> result::Result<T, E>,
+        F: FnMut(&Row<'_>) -> result::Result<T, E>,
     {
         let rows = self.query_named(params)?;
         Ok(AndThenRows::new(rows, f))
@@ -389,7 +388,7 @@ impl Statement<'_> {
         P::Item: ToSql,
     {
         let mut rows = self.query(params)?;
-        let exists = rows.next().is_some();
+        let exists = rows.next()?.is_some();
         Ok(exists)
     }
 
@@ -410,7 +409,7 @@ impl Statement<'_> {
     where
         P: IntoIterator,
         P::Item: ToSql,
-        F: FnOnce(&Row<'_, '_>) -> Result<T>,
+        F: FnOnce(&Row<'_>) -> Result<T>,
     {
         let mut rows = self.query(params)?;
 
@@ -507,13 +506,7 @@ impl Statement<'_> {
             ValueRef::Real(r) => unsafe { ffi::sqlite3_bind_double(ptr, col as c_int, r) },
             ValueRef::Text(s) => unsafe {
                 let (c_str, len, destructor) = str_for_sqlite(s)?;
-                ffi::sqlite3_bind_text(
+                ffi::sqlite3_bind_text(ptr, col as c_int, c_str, len, destructor)
-                    ptr,
-                    col as c_int,
-                    c_str,
-                    len,
-                    destructor,
-                )
             },
             ValueRef::Blob(b) => unsafe {
                 let length = len_as_c_int(b.len())?;
@@ -786,7 +779,7 @@ mod test {
             .unwrap();
         let mut rows = stmt.query_named(&[(":name", &"one")]).unwrap();
 
-        let id: Result<i32> = rows.next().unwrap().and_then(|row| row.get(0));
+        let id: Result<i32> = rows.next().unwrap().unwrap().get(0);
         assert_eq!(Ok(1), id);
     }
 

src/types/from_sql.rs

@@ -151,7 +151,7 @@ impl FromSql for bool {
 
 impl FromSql for String {
     fn column_result(value: ValueRef<'_>) -> FromSqlResult<Self> {
-        value.as_str().map(|s| s.to_string())
+        value.as_str().map(ToString::to_string)
     }
 }
 

src/vtab/csvtab.rs

@@ -345,6 +345,7 @@ impl From<csv::Error> for Error {
 
 #[cfg(test)]
 mod test {
+    use fallible_iterator::FallibleIterator;
     use crate::vtab::csvtab;
     use crate::{Connection, Result, NO_PARAMS};
 
@@ -363,8 +364,7 @@ mod test {
             }
 
             let ids: Result<Vec<i32>> = s
-                .query_map(NO_PARAMS, |row| row.get::<_, i32>(0))
+                .query(NO_PARAMS).unwrap().map(|row| row.get::<_, i32>(0))
-                .unwrap()
                 .collect();
             let sum = ids.unwrap().iter().sum::<i32>();
             assert_eq!(sum, 15);