cassandra-rs non-idiomatic use of iterators leads to use after free

@@ -8,7 +8,7 @@ keywords = [ "Cassandra", "binding", "CQL", "client", "database" ]
 categories = [ "api-bindings", "database", "external-ffi-bindings", "asynchronous" ]
 license = "Apache-2.0"
 name = "cassandra-cpp"
-version = "2.0.2-pre"
+version = "3.0.0"
 authors = ["Tupshin Harper <tupshin@tupshin.com>", "Keith Wansbrough <Keith.Wansbrough@metaswitch.com>"]
 edition = "2018"
 

@@ -14,6 +14,34 @@ version number is tracked in the file `VERSION`.
 
 ### Fixed
 
+## [3.0.0]
+
+### Added
+- `get_decimal()` now gets a Cassanda decimal value as `BigDecimal`,
+  `bind_decimal()` does the reverse.
+
+### Changed
+- Date type is now `u32`.
+
+### Fixed
+- Soundness issue fixed - the various iterators no longer implement
+  `std::iter::Iterator`, but instead a new `LendingIterator` trait.
+  For migration guide, see [`README`](README.md#lending-iterator-api-version-30).
+- Many types now take a lifetime argument, e.g., `Value` is now `Value<'a>`,
+  `ResultIterator` is now `ResultIterator<'a>`.
+- `RowIterator` no longer implements `Display` (since it would consume the
+  iterator); however `Row` does.
+- `Result::into_iter()` and `Row::into_iter()` are now both `iter()`, since they
+  do not consume their argument.
+- `TupleIterator` is removed - it was never used, since you use the set iterator
+  (Value::get_set()) for lists, sets, and tuples.
+- `ConstDataType::sub_data_by_name` and `ConstDataType::sub_type_name` now take
+  `&self` rather than an explicit argument.
+- `FunctionMeta::argument` now returns the name and type, rather than just `()`.
+- The underlying raw pointer is no longer exposed as a field of `Cluster`,
+  `ResultIterator`, and `RowIterator`.
+- Many types now implement `Sync` as well as `Send`.
+
 ## [2.0.1] - 2023-06-08
 
 ### Fixed

@@ -26,26 +26,35 @@ async fn do_work(session: &Session) -> Result<()> {
     let result = session.execute("SELECT * FROM testks.user;").await?;
 
     println!("Overall result: {}", result);
-    for row in result.iter() {
+    let mut iter = result.iter();
+    while let Some(row) = iter.next() {
         println!("Row: {}", row);
 
         let first_name: String = row.get_by_name("first_name")?;
         let addresses: HashMap<String, String> = {
             let maybe_iter: Result<MapIterator> = row.get_by_name("addresses");
             match maybe_iter {
                 Err(_) => HashMap::new(),
-                Ok(addresses_iter) => addresses_iter
-                    .map(|(k, v)| Ok((k.get_string()?, v.get_string()?)))
-                    .collect::<Result<_>>()?,
+                Ok(mut addresses_iter) => {
+                    let mut map = HashMap::new();
+                    while let Some((k, v)) = addresses_iter.next() {
+                        map.insert(k.get_string()?, v.get_string()?);
+                    }
+                    map
+                }
             }
         };
-        let emails_iter: SetIterator = row.get_by_name("email")?;
-        let emails: Vec<String> = emails_iter.map(|v| v.get_string()).collect::<Result<_>>()?;
+        let mut emails_iter: SetIterator = row.get_by_name("email")?;
+        let mut emails: Vec<String> = vec![];
+        while let Some(v) = emails_iter.next() {
+            emails.push(v.get_string()?);
+        }
         let last_name: String = row.get_by_name("last_name")?;
-        let phone_numbers_iter: SetIterator = row.get_by_name("phone_numbers")?;
-        let phone_numbers: Vec<String> = phone_numbers_iter
-            .map(|v| v.get_string())
-            .collect::<Result<_>>()?;
+        let mut phone_numbers_iter: SetIterator = row.get_by_name("phone_numbers")?;
+        let mut phone_numbers: Vec<String> = vec![];
+        while let Some(v) = phone_numbers_iter.next() {
+            phone_numbers.push(v.get_string()?);
+        }
         let title: i32 = row.get_by_name("title")?;
         println!(
             " == {} {:?} {:?} {} {:?} {}",

@@ -12,7 +12,8 @@ async fn main() -> Result<()> {
         .await?;
     println!("{}", result);
 
-    for row in result.iter() {
+    let mut iter = result.iter();
+    while let Some(row) = iter.next() {
         let col: String = row.get_by_name("keyspace_name")?;
         println!("ks name = {}", col);
     }

@@ -25,7 +25,8 @@ async fn main() -> Result<()> {
         .unwrap();
 
     println!("{}", result);
-    for row in result.iter() {
+    let mut iter = result.iter();
+    while let Some(row) = iter.next() {
         let col: String = row.get_by_name(col_name).unwrap();
         println!("ks name = {}", col);
     }

@@ -55,6 +55,56 @@ For a straightforward example see [`simple.rs`](examples/simple.rs).
 There are additional examples included with the project in [`tests`](tests/) and
 [`examples`](examples/).
 
+## Lending iterator API (version 3.0)
+
+Version 3.0 fixes a soundness issue with the previous API. The iterators in the
+underlying Cassandra driver invalidate the current item when `next()` is called,
+and this was not reflected in the Rust binding prior to version 3.
+
+To deal with this, the various iterators (`ResultIterator`, `RowIterator`,
+`MapIterator`, `SetIterator`, `FieldIterator`, `UserTypeIterator`,
+`KeyspaceIterator`, `FunctionIterator`, `AggregateIterator`, `TableIterator`,
+`ColumnIterator`) no longer implement `std::iter::Iterator`. Instead, since this
+is a [lending
+iterator,](https://blog.rust-lang.org/2022/11/03/Rust-1.65.0.html#generic-associated-types-gats)
+these types all implement a new `LendingIterator` trait. We define this
+ourselves because there is currently no widely-used crate that implements it.
+
+To upgrade, change
+
+```rust
+for row in result {
+  // ... do something with row ...
+}
+```
+
+to
+
+```rust
+let mut iter = result.iter();
+while let Some(row) = iter.next() {
+  // ... do something with row ...
+}
+```
+
+The intermediate variable `iter` is necessary, otherwise you will infinitely
+visit the first row of the result!
+
+Other changes:
+
+* Many types now take a lifetime argument, e.g., `Value` is now `Value<'a>`,
+  `ResultIterator` is now `ResultIterator<'a>`. In almost all cases you can omit
+  this and it will be inferred for you. If not, you can usually write
+  `Value<'_>` to let Rust worry about it for you.
+* `RowIterator` no longer implements `Display` (since it would consume the
+  iterator); however `Row` does.
+* `TupleIterator` is removed - it was never used, since you use the set iterator
+  (Value::get_set()) for lists, sets, and tuples.
+* `ConstDataType::sub_data_by_name` and `ConstDataType::sub_type_name` now take
+  `&self` rather than an explicit argument.
+* `FunctionMeta::argument` now returns the name and type, rather than just `()`.
+
+
 ## New session API (version 2.0)
 
 Version 2.0 introduces a new and safer API. `Statement`s (and

@@ -18,10 +18,6 @@ use crate::cassandra_sys::cass_batch_set_serial_consistency;
 use crate::cassandra_sys::cass_batch_set_timestamp;
 use crate::cassandra_sys::CassBatch as _Batch;
 use crate::cassandra_sys::CassBatchType_;
-use crate::cassandra_sys::CassConsistency;
-use crate::cassandra_sys::CassCustomPayload as _CassCustomPayload;
-use std::ffi::NulError;
-use std::os::raw::c_char;
 
 #[derive(Debug)]
 struct BatchInner(*mut _Batch);
@@ -31,9 +27,10 @@ struct BatchInner(*mut _Batch);
 #[derive(Debug)]
 pub struct Batch(BatchInner, Session);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
 unsafe impl Send for BatchInner {}
+unsafe impl Sync for BatchInner {}
 
 impl ProtectedInner<*mut _Batch> for BatchInner {
     #[inline(always)]

@@ -48,23 +48,15 @@ use crate::cassandra_sys::cass_cluster_set_whitelist_filtering;
 use crate::cassandra_sys::cass_cluster_set_write_bytes_high_water_mark;
 use crate::cassandra_sys::cass_cluster_set_write_bytes_low_water_mark;
 use crate::cassandra_sys::cass_false;
-use crate::cassandra_sys::cass_future_error_code;
+
 use crate::cassandra_sys::cass_session_connect;
 use crate::cassandra_sys::cass_session_connect_keyspace_n;
-use crate::cassandra_sys::cass_session_new;
+
 use crate::cassandra_sys::cass_true;
 use crate::cassandra_sys::CassCluster as _Cluster;
 
-use std::ffi::CStr;
-use std::ffi::NulError;
-use std::fmt;
-use std::fmt::Display;
-use std::iter::Map;
-use std::net::AddrParseError;
-use std::net::Ipv4Addr;
 use std::os::raw::c_char;
-use std::result;
-use std::str::FromStr;
+
 use std::time::Duration;
 
 /// A CQL protocol version is just an integer.
@@ -85,11 +77,12 @@ pub type CqlProtocol = i32;
 /// # }
 /// ```
 #[derive(Debug)]
-pub struct Cluster(pub *mut _Cluster);
+pub struct Cluster(*mut _Cluster);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
 unsafe impl Send for Cluster {}
+unsafe impl Sync for Cluster {}
 
 impl Drop for Cluster {
     /// Frees a cluster instance.

@@ -10,7 +10,7 @@ use crate::cassandra::uuid::Uuid;
 use crate::cassandra_sys::cass_collection_append_bool;
 use crate::cassandra_sys::cass_collection_append_bytes;
 use crate::cassandra_sys::cass_collection_append_collection;
-use crate::cassandra_sys::cass_collection_append_decimal;
+
 use crate::cassandra_sys::cass_collection_append_double;
 use crate::cassandra_sys::cass_collection_append_float;
 use crate::cassandra_sys::cass_collection_append_inet;
@@ -124,9 +124,10 @@ pub trait CassCollection: Sized {
 #[derive(Debug)]
 pub struct List(*mut _CassCollection);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
 unsafe impl Send for List {}
+unsafe impl Sync for List {}
 
 impl ProtectedInner<*mut _CassCollection> for List {
     fn inner(&self) -> *mut _CassCollection {
@@ -302,9 +303,10 @@ impl CassCollection for List {
 #[derive(Debug)]
 pub struct Set(*mut _CassCollection);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
 unsafe impl Send for Set {}
+unsafe impl Sync for Set {}
 
 impl Drop for Set {
     fn drop(&mut self) {
@@ -438,9 +440,10 @@ impl CassCollection for Set {
 #[derive(Debug)]
 pub struct Map(*mut _CassCollection);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
 unsafe impl Send for Map {}
+unsafe impl Sync for Map {}
 
 impl Drop for Map {
     fn drop(&mut self) {

@@ -1,10 +1,5 @@
-use crate::cassandra::util::Protected;
 use crate::cassandra_sys::CassConsistency_;
 
-use crate::cassandra_sys::cass_consistency_string;
-
-use std::ffi::CStr;
-
 /// A Cassandra consistency level.
 #[derive(Debug, Eq, PartialEq, Copy, Clone, Hash)]
 #[allow(missing_docs)] // Meanings are defined in CQL documentation.

@@ -1,9 +1,9 @@
 use crate::cassandra::error::*;
-use crate::cassandra::util::{Protected, ProtectedInner, ProtectedWithSession};
+use crate::cassandra::util::{Protected, ProtectedInner};
 
 use crate::cassandra_sys::cass_custom_payload_free;
 use crate::cassandra_sys::cass_custom_payload_new;
-use crate::cassandra_sys::cass_custom_payload_set;
+
 use crate::cassandra_sys::cass_custom_payload_set_n;
 use crate::cassandra_sys::CassCustomPayload as _CassCustomPayload;
 use std::collections::HashMap;
@@ -15,9 +15,10 @@ pub type CustomPayloadResponse = HashMap<String, Vec<u8>>;
 #[derive(Debug)]
 pub struct CustomPayload(*mut _CassCustomPayload);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
 unsafe impl Send for CustomPayload {}
+unsafe impl Sync for CustomPayload {}
 
 impl ProtectedInner<*mut _CassCustomPayload> for CustomPayload {
     #[inline(always)]

@@ -27,18 +27,25 @@ use crate::cassandra_sys::cass_user_type_new_from_data_type;
 use crate::cassandra_sys::CassDataType as _CassDataType;
 
 use std::ffi::CString;
+use std::marker::PhantomData;
 use std::os::raw::c_char;
 
-/// Any cassandra datatype
+/// Any Cassandra datatype. This is an owned type.
 #[derive(Debug)]
 pub struct DataType(*mut _CassDataType);
+
+/// Any Cassandra datatype. This is a reference type.
+//
+// Borrowed from whatever descriptor contains the type, e.g., a `SchemaMeta`.
 #[derive(Debug)]
-pub struct ConstDataType(*const _CassDataType);
+pub struct ConstDataType<'a>(*const _CassDataType, PhantomData<&'a _CassDataType>);
 
-// The underlying C types have no thread-local state, but do not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
+// The underlying C types have no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
 unsafe impl Send for DataType {}
-unsafe impl Send for ConstDataType {}
+unsafe impl Sync for DataType {}
+unsafe impl Send for ConstDataType<'_> {}
+unsafe impl Sync for ConstDataType<'_> {}
 
 impl ProtectedInner<*mut _CassDataType> for DataType {
     fn inner(&self) -> *mut _CassDataType {
@@ -55,18 +62,18 @@ impl Protected<*mut _CassDataType> for DataType {
     }
 }
 
-impl ProtectedInner<*const _CassDataType> for ConstDataType {
+impl ProtectedInner<*const _CassDataType> for ConstDataType<'_> {
     fn inner(&self) -> *const _CassDataType {
         self.0
     }
 }
 
-impl Protected<*const _CassDataType> for ConstDataType {
+impl Protected<*const _CassDataType> for ConstDataType<'_> {
     fn build(inner: *const _CassDataType) -> Self {
         if inner.is_null() {
             panic!("Unexpected null pointer")
         };
-        ConstDataType(inner)
+        ConstDataType(inner, PhantomData)
     }
 }
 
@@ -77,7 +84,7 @@ impl Drop for DataType {
     }
 }
 
-impl ConstDataType {
+impl ConstDataType<'_> {
     /// Creates a new user defined type from existing data type.
     pub fn new_user_type(&self) -> UserType {
         unsafe { UserType::build(cass_user_type_new_from_data_type(self.0)) }
@@ -233,7 +240,7 @@ impl DataType {
     /// Gets the sub-data type of a UDT (user defined type) at the specified index.
     ///
     /// <b>Note:</b> Only valid for UDT data types.
-    pub fn sub_data_type_by_name<S>(data_type: DataType, name: S) -> ConstDataType
+    pub fn sub_data_type_by_name<S>(&self, name: S) -> ConstDataType
     where
         S: Into<String>,
     {
@@ -242,7 +249,7 @@ impl DataType {
             let name_ptr = name_str.as_ptr() as *const c_char;
             // TODO: can return NULL
             ConstDataType::build(cass_data_type_sub_data_type_by_name_n(
-                data_type.0,
+                self.0,
                 name_ptr,
                 name_str.len(),
             ))
@@ -252,14 +259,14 @@ impl DataType {
     /// Gets the sub-type name of a UDT (user defined type) at the specified index.
     ///
     /// <b>Note:</b> Only valid for UDT data types.
-    pub fn sub_type_name<S>(data_type: DataType, index: usize, name: S) -> Result<()>
+    pub fn sub_type_name<S>(&self, index: usize, name: S) -> Result<()>
     where
         S: Into<String>,
     {
         unsafe {
             let name2 = CString::new(name.into())?;
             cass_data_type_sub_type_name(
-                data_type.0,
+                self.0,
                 index,
                 &mut name2.as_ptr(),
                 &mut (name2.as_bytes().len()),

@@ -4,9 +4,10 @@ use crate::cassandra::value::ValueType;
 use crate::cassandra::write_type::WriteType;
 
 use crate::cassandra_sys::cass_error_desc;
-use crate::cassandra_sys::cass_error_result_code;
+
 use crate::cassandra_sys::cass_error_result_free;
-use crate::cassandra_sys::cass_future_error_code;
+
+use crate::cassandra_sys::cass_false;
 use crate::cassandra_sys::cass_future_error_message;
 use crate::cassandra_sys::cass_future_get_error_result;
 use crate::cassandra_sys::CassErrorResult as CassErrorResult_;
@@ -19,14 +20,12 @@ use crate::cassandra_sys::{
     cass_error_result_num_failures, cass_error_result_responses_received,
     cass_error_result_responses_required, cass_error_result_table, cass_error_result_write_type,
 };
-use crate::cassandra_sys::{cass_false, cass_true};
 use crate::Session;
 
-use std::error::Error as IError;
-use std::ffi::{CStr, CString};
-use std::fmt::{Debug, Display, Formatter};
-use std::os::raw::c_char;
-use std::{fmt, mem, slice, str};
+use std::ffi::CStr;
+use std::fmt::Debug;
+
+use std::{slice, str};
 
 // Define the errors that may be returned by this driver.
 error_chain! {

@@ -1,65 +1,11 @@
-use crate::cassandra::inet::Inet;
-use crate::cassandra::iterator::MapIterator;
-use crate::cassandra::iterator::SetIterator;
 // use decimal::d128;
-use crate::cassandra::error::*;
-use crate::cassandra::util::Protected;
-use crate::cassandra::uuid::Uuid;
-use crate::cassandra::value::{write_map, write_set, Value, ValueType};
 
-use crate::cassandra_sys::cass_iterator_from_collection;
-use crate::cassandra_sys::cass_iterator_from_map;
-use crate::cassandra_sys::cass_true;
-use crate::cassandra_sys::cass_value_get_bool;
-use crate::cassandra_sys::cass_value_get_decimal;
-use crate::cassandra_sys::cass_value_get_double;
-use crate::cassandra_sys::cass_value_get_float;
-use crate::cassandra_sys::cass_value_get_inet;
-use crate::cassandra_sys::cass_value_get_int16;
-use crate::cassandra_sys::cass_value_get_int32;
-use crate::cassandra_sys::cass_value_get_int64;
-use crate::cassandra_sys::cass_value_get_int8;
-use crate::cassandra_sys::cass_value_get_string;
-use crate::cassandra_sys::cass_value_get_uint32;
-use crate::cassandra_sys::cass_value_get_uuid;
-use crate::cassandra_sys::cass_value_type;
-use crate::cassandra_sys::CASS_ERROR_LIB_INVALID_VALUE_TYPE;
-use crate::cassandra_sys::CASS_VALUE_TYPE_ASCII;
-use crate::cassandra_sys::CASS_VALUE_TYPE_BIGINT;
-use crate::cassandra_sys::CASS_VALUE_TYPE_BLOB;
-use crate::cassandra_sys::CASS_VALUE_TYPE_BOOLEAN;
-use crate::cassandra_sys::CASS_VALUE_TYPE_COUNTER;
-use crate::cassandra_sys::CASS_VALUE_TYPE_CUSTOM;
-use crate::cassandra_sys::CASS_VALUE_TYPE_DATE;
-use crate::cassandra_sys::CASS_VALUE_TYPE_DECIMAL;
-use crate::cassandra_sys::CASS_VALUE_TYPE_DOUBLE;
-use crate::cassandra_sys::CASS_VALUE_TYPE_FLOAT;
-use crate::cassandra_sys::CASS_VALUE_TYPE_INET;
-use crate::cassandra_sys::CASS_VALUE_TYPE_INT;
-use crate::cassandra_sys::CASS_VALUE_TYPE_LAST_ENTRY;
-use crate::cassandra_sys::CASS_VALUE_TYPE_LIST;
-use crate::cassandra_sys::CASS_VALUE_TYPE_MAP;
-use crate::cassandra_sys::CASS_VALUE_TYPE_SET;
-use crate::cassandra_sys::CASS_VALUE_TYPE_SMALL_INT;
-use crate::cassandra_sys::CASS_VALUE_TYPE_TEXT;
-use crate::cassandra_sys::CASS_VALUE_TYPE_TIME;
-use crate::cassandra_sys::CASS_VALUE_TYPE_TIMESTAMP;
-use crate::cassandra_sys::CASS_VALUE_TYPE_TIMEUUID;
-use crate::cassandra_sys::CASS_VALUE_TYPE_TINY_INT;
-use crate::cassandra_sys::CASS_VALUE_TYPE_TUPLE;
-use crate::cassandra_sys::CASS_VALUE_TYPE_UDT;
-use crate::cassandra_sys::CASS_VALUE_TYPE_UNKNOWN;
-use crate::cassandra_sys::CASS_VALUE_TYPE_UUID;
-use crate::cassandra_sys::CASS_VALUE_TYPE_VARCHAR;
-use crate::cassandra_sys::CASS_VALUE_TYPE_VARINT;
+use crate::cassandra::value::{Value, ValueType};
 
 use std::fmt;
 use std::fmt::Debug;
 use std::fmt::Display;
 use std::fmt::Formatter;
-use std::mem;
-use std::slice;
-use std::str;
 
 // #[repr(C)]
 // #[derive(Copy,Debug,Clone)]
@@ -88,26 +34,28 @@ use std::str;
 // }
 
 /// A field's metadata
-pub struct Field {
+//
+// Borrowed from wherever the value is borrowed from.
+pub struct Field<'a> {
     /// The field's name
     pub name: String,
     /// The field's value
-    pub value: Value,
+    pub value: Value<'a>,
 }
 
-impl Debug for Field {
+impl Debug for Field<'_> {
     fn fmt(&self, f: &mut Formatter) -> fmt::Result {
         write!(f, "{} Cassandra type", self.get_type())
     }
 }
 
-impl Display for Field {
+impl Display for Field<'_> {
     fn fmt(&self, f: &mut Formatter) -> fmt::Result {
         write!(f, "{} Cassandra type", self.get_type())
     }
 }
 
-impl Field {
+impl<'a> Field<'a> {
     /// Gets the name of this field
     pub fn get_name(&self) -> String {
         self.name.clone()
@@ -119,7 +67,7 @@ impl Field {
     }
 
     /// Gets the value of this field
-    pub fn get_value(&self) -> &Value {
+    pub fn get_value(&self) -> &Value<'a> {
         &self.value
     }
 }

@@ -1,28 +1,27 @@
-use crate::cassandra::custom_payload::{CustomPayload, CustomPayloadResponse};
+use crate::cassandra::custom_payload::CustomPayloadResponse;
 use crate::cassandra::error::*;
 use crate::cassandra::prepared::PreparedStatement;
 use crate::cassandra::result::CassResult;
 use crate::cassandra::util::{Protected, ProtectedWithSession};
-use crate::cassandra::write_type::WriteType;
+
 use crate::cassandra_sys::cass_future_custom_payload_item;
 use crate::cassandra_sys::cass_future_custom_payload_item_count;
 use crate::cassandra_sys::cass_future_error_code;
-use crate::cassandra_sys::cass_future_error_message;
+
 use crate::cassandra_sys::cass_future_free;
-use crate::cassandra_sys::cass_future_get_error_result;
+
 use crate::cassandra_sys::cass_future_get_prepared;
 use crate::cassandra_sys::cass_future_get_result;
 use crate::cassandra_sys::cass_future_ready;
 use crate::cassandra_sys::cass_future_set_callback;
-use crate::cassandra_sys::CassError_;
+
+use crate::cassandra_sys::cass_true;
 use crate::cassandra_sys::CassFuture as _Future;
 use crate::cassandra_sys::CASS_OK;
-use crate::cassandra_sys::{cass_false, cass_true};
-use crate::{cassandra::consistency::Consistency, Session};
+use crate::Session;
 
 use parking_lot::Mutex;
 
-use std::collections::HashMap;
 use std::future::Future;
 use std::marker::PhantomData;
 use std::mem;

@@ -9,7 +9,7 @@ use std::default::Default;
 use std::ffi::CStr;
 use std::fmt;
 use std::fmt::{Debug, Formatter};
-use std::mem;
+
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
 use std::os::raw::c_char;
 use std::str::FromStr;

@@ -1,4 +1,3 @@
-use crate::cassandra::data_type::DataType;
 use crate::cassandra::error::*;
 use crate::cassandra::field::Field;
 use crate::cassandra::schema::aggregate_meta::AggregateMeta;
@@ -28,25 +27,78 @@ use crate::cassandra_sys::cass_iterator_get_user_type_field_value;
 use crate::cassandra_sys::cass_iterator_get_value;
 use crate::cassandra_sys::cass_iterator_next;
 use crate::cassandra_sys::cass_true;
-use std::{mem, slice, str};
-
-/// Iterates over the  aggregate metadata entries(??)
+use crate::cassandra_sys::CassKeyspaceMeta;
+use crate::cassandra_sys::CassSchemaMeta;
+use crate::cassandra_sys::CassTableMeta;
+use crate::cassandra_sys::CassValue as _CassValue;
+
+use std::marker::PhantomData;
+use std::{slice, str};
+
+/// Iterator that only allows access to a single item at a time. You must stop
+/// using the returned item before getting the next.
+///
+/// Ultimately we will move to use a common crate for this, but to date
+/// there is no good crate to follow.
+/// https://blog.rust-lang.org/2022/11/03/Rust-1.65.0.html#generic-associated-types-gats
+/// and https://github.com/Crazytieguy/gat-lending-iterator were references
+/// for this code.
+///
+/// The idiomatic way to work with this trait is as follows:
+///
+/// ```
+/// # use cassandra_cpp::*;
+/// # struct MyLI;
+/// # impl LendingIterator for MyLI {
+/// #   type Item<'a> = ();
+/// #   fn next<'a>(&'a mut self) -> Option<Self::Item<'a>> { None }
+/// #   fn size_hint(&self) -> (usize, Option<usize>) { (0, Some(0)) }
+/// # }
+/// # let mut lending_iterator = MyLI;
+/// while let Some(row) = lending_iterator.next() {
+///   // ... do something with `row` ...
+/// }
+/// ```
+pub trait LendingIterator {
+    /// The type of each item.
+    type Item<'a>
+    where
+        Self: 'a;
+
+    /// Retrieve the next item from the iterator; it lives only as long as the
+    /// mutable reference to the iterator.
+    fn next(&mut self) -> Option<Self::Item<'_>>;
+
+    /// Minimum and optional maximum expected length of the iterator.
+    /// Default implementation returns `(0, None)`.
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (0, None)
+    }
+}
+
+/// Iterator over the aggregates in the keyspace.
+///
+/// A Cassandra iterator is a `LendingIterator` because it borrows from some
+/// underlying value, but owns a single item. Each time `next()` is invoked it
+/// decodes the current item into that item, thus invalidating its previous
+/// value.
 #[derive(Debug)]
-pub struct AggregateIterator(*mut _CassIterator);
+pub struct AggregateIterator<'a>(*mut _CassIterator, PhantomData<&'a CassKeyspaceMeta>);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
-unsafe impl Send for AggregateIterator {}
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
+unsafe impl Send for AggregateIterator<'_> {}
+unsafe impl Sync for AggregateIterator<'_> {}
 
-impl Drop for AggregateIterator {
+impl Drop for AggregateIterator<'_> {
     fn drop(&mut self) {
         unsafe { cass_iterator_free(self.0) }
     }
 }
 
-impl Iterator for AggregateIterator {
-    type Item = AggregateMeta;
-    fn next(&mut self) -> Option<<Self as Iterator>::Item> {
+impl LendingIterator for AggregateIterator<'_> {
+    type Item<'a> = AggregateMeta<'a> where Self: 'a;
+    fn next(&mut self) -> Option<<Self as LendingIterator>::Item<'_>> {
         unsafe {
             match cass_iterator_next(self.0) {
                 cass_false => None,
@@ -59,23 +111,29 @@ impl Iterator for AggregateIterator {
     }
 }
 
-/// Iterater over the fields of a UDT
+/// Iterator over the fields of a UDT
+///
+/// A Cassandra iterator is a `LendingIterator` because it borrows from some
+/// underlying value, but owns a single item. Each time `next()` is invoked it
+/// decodes the current item into that item, thus invalidating its previous
+/// value.
 #[derive(Debug)]
-pub struct UserTypeIterator(*mut _CassIterator);
+pub struct UserTypeIterator<'a>(*mut _CassIterator, PhantomData<&'a _CassValue>);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
-unsafe impl Send for UserTypeIterator {}
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
+unsafe impl Send for UserTypeIterator<'_> {}
+unsafe impl Sync for UserTypeIterator<'_> {}
 
-impl Drop for UserTypeIterator {
+impl Drop for UserTypeIterator<'_> {
     fn drop(&mut self) {
         unsafe { cass_iterator_free(self.0) }
     }
 }
 
-impl Iterator for UserTypeIterator {
-    type Item = (String, Value);
-    fn next(&mut self) -> Option<<Self as Iterator>::Item> {
+impl LendingIterator for UserTypeIterator<'_> {
+    type Item<'a> = (String, Value<'a>) where Self: 'a;
+    fn next(&mut self) -> Option<<Self as LendingIterator>::Item<'_>> {
         unsafe {
             match cass_iterator_next(self.0) {
                 cass_false => None,
@@ -85,8 +143,8 @@ impl Iterator for UserTypeIterator {
     }
 }
 
-impl UserTypeIterator {
-    fn get_field_name(&mut self) -> String {
+impl UserTypeIterator<'_> {
+    fn get_field_name(&self) -> String {
         unsafe {
             let mut name = std::ptr::null();
             let mut name_length = 0;
@@ -100,22 +158,29 @@ impl UserTypeIterator {
                 .expect("Cassandra error during iteration")
         }
     }
-    fn get_field_value(&mut self) -> Value {
+
+    fn get_field_value(&self) -> Value {
         unsafe { Value::build(cass_iterator_get_user_type_field_value(self.0)) }
     }
 }
 
-/// Iterater over the  function metadata entries(??)
+/// Iterator over the functions in a keyspace.
+///
+/// A Cassandra iterator is a `LendingIterator` because it borrows from some
+/// underlying value, but owns a single item. Each time `next()` is invoked it
+/// decodes the current item into that item, thus invalidating its previous
+/// value.
 #[derive(Debug)]
-pub struct FunctionIterator(*mut _CassIterator);
+pub struct FunctionIterator<'a>(*mut _CassIterator, PhantomData<&'a CassKeyspaceMeta>);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
-unsafe impl Send for FunctionIterator {}
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
+unsafe impl Send for FunctionIterator<'_> {}
+unsafe impl Sync for FunctionIterator<'_> {}
 
-impl Iterator for FunctionIterator {
-    type Item = FunctionMeta;
-    fn next(&mut self) -> Option<<Self as Iterator>::Item> {
+impl LendingIterator for FunctionIterator<'_> {
+    type Item<'a> = FunctionMeta<'a> where Self: 'a;
+    fn next(&mut self) -> Option<<Self as LendingIterator>::Item<'_>> {
         unsafe {
             match cass_iterator_next(self.0) {
                 cass_false => None,
@@ -125,17 +190,23 @@ impl Iterator for FunctionIterator {
     }
 }
 
-/// Iterater over the table's metadata entries(??)
+/// Iterator over the tables in a keyspace.
+///
+/// A Cassandra iterator is a `LendingIterator` because it borrows from some
+/// underlying value, but owns a single item. Each time `next()` is invoked it
+/// decodes the current item into that item, thus invalidating its previous
+/// value.
 #[derive(Debug)]
-pub struct TableIterator(*mut _CassIterator);
+pub struct TableIterator<'a>(*mut _CassIterator, PhantomData<&'a CassKeyspaceMeta>);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
-unsafe impl Send for TableIterator {}
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
+unsafe impl Send for TableIterator<'_> {}
+unsafe impl Sync for TableIterator<'_> {}
 
-impl Iterator for TableIterator {
-    type Item = TableMeta;
-    fn next(&mut self) -> Option<<Self as Iterator>::Item> {
+impl LendingIterator for TableIterator<'_> {
+    type Item<'a> = TableMeta<'a> where Self: 'a;
+    fn next(&mut self) -> Option<<Self as LendingIterator>::Item<'_>> {
         unsafe {
             match cass_iterator_next(self.0) {
                 cass_false => None,
@@ -145,17 +216,23 @@ impl Iterator for TableIterator {
     }
 }
 
-/// Iterater over the keyspace's metadata entries(??)
+/// Iterator over the keyspaces in the schema.
+///
+/// A Cassandra iterator is a `LendingIterator` because it borrows from some
+/// underlying value, but owns a single item. Each time `next()` is invoked it
+/// decodes the current item into that item, thus invalidating its previous
+/// value.
 #[derive(Debug)]
-pub struct KeyspaceIterator(*mut _CassIterator);
+pub struct KeyspaceIterator<'a>(*mut _CassIterator, PhantomData<&'a CassSchemaMeta>);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
-unsafe impl Send for KeyspaceIterator {}
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
+unsafe impl Send for KeyspaceIterator<'_> {}
+unsafe impl Sync for KeyspaceIterator<'_> {}
 
-impl Iterator for KeyspaceIterator {
-    type Item = KeyspaceMeta;
-    fn next(&mut self) -> Option<<Self as Iterator>::Item> {
+impl LendingIterator for KeyspaceIterator<'_> {
+    type Item<'a> = KeyspaceMeta<'a> where Self: 'a;
+    fn next(&mut self) -> Option<<Self as LendingIterator>::Item<'_>> {
         unsafe {
             match cass_iterator_next(self.0) {
                 cass_false => None,
@@ -165,17 +242,23 @@ impl Iterator for KeyspaceIterator {
     }
 }
 
-/// Iterater over the columns's metadata entries(??)
+/// Iterator over the columns in a table.
+///
+/// A Cassandra iterator is a `LendingIterator` because it borrows from some
+/// underlying value, but owns a single item. Each time `next()` is invoked it
+/// decodes the current item into that item, thus invalidating its previous
+/// value.
 #[derive(Debug)]
-pub struct ColumnIterator(*mut _CassIterator);
+pub struct ColumnIterator<'a>(*mut _CassIterator, PhantomData<&'a CassTableMeta>);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
-unsafe impl Send for ColumnIterator {}
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
+unsafe impl Send for ColumnIterator<'_> {}
+unsafe impl Sync for ColumnIterator<'_> {}
 
-impl Iterator for ColumnIterator {
-    type Item = ColumnMeta;
-    fn next(&mut self) -> Option<<Self as Iterator>::Item> {
+impl LendingIterator for ColumnIterator<'_> {
+    type Item<'a> = ColumnMeta<'a> where Self: 'a;
+    fn next(&mut self) -> Option<<Self as LendingIterator>::Item<'_>> {
         unsafe {
             match cass_iterator_next(self.0) {
                 cass_false => None,
@@ -185,17 +268,27 @@ impl Iterator for ColumnIterator {
     }
 }
 
-/// Iterater over the field's metadata entries(??)
+/// Iterator over the fields in a metadata object.
+///
+/// A Cassandra iterator is a `LendingIterator` because it borrows from some
+/// underlying value, but owns a single item. Each time `next()` is invoked it
+/// decodes the current item into that item, thus invalidating its previous
+/// value.
+//
+// Could be one of several underlying types; CassTableMeta is just a
+// representative. Since it's a phantom, it doesn't matter which type we name.
 #[derive(Debug)]
-pub struct FieldIterator(*mut _CassIterator);
+pub struct FieldIterator<'a>(*mut _CassIterator, PhantomData<&'a CassTableMeta>);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
-unsafe impl Send for FieldIterator {}
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
+unsafe impl Send for FieldIterator<'_> {}
+unsafe impl Sync for FieldIterator<'_> {}
 
-impl Iterator for FieldIterator {
-    type Item = Field;
-    fn next(&mut self) -> Option<<Self as Iterator>::Item> {
+impl LendingIterator for FieldIterator<'_> {
+    type Item<'a> = Field<'a> where Self: 'a;
+
+    fn next(&mut self) -> Option<<Self as LendingIterator>::Item<'_>> {
         unsafe {
             match cass_iterator_next(self.0) {
                 cass_false => None,
@@ -217,182 +310,164 @@ impl Iterator for FieldIterator {
     }
 }
 
-// pub struct CassIteratorType(_CassIteratorType);
-
-// impl CassIteratorType {
-//    pub fn new(_type: _CassIteratorType) -> Self { CassIteratorType(_type) }
-// }
-
-// impl Protected<*mut _Batch> for CassIterator {
-//    fn inner(&self) -> *mut _CassIterator {
-//        self.0
-//    }
-//    fn build(inner: *mut _CassIterator) -> Self {
-//        CassIterator(inner)
-//    }
-// }
-
-impl ProtectedInner<*mut _CassIterator> for UserTypeIterator {
+impl ProtectedInner<*mut _CassIterator> for UserTypeIterator<'_> {
     fn inner(&self) -> *mut _CassIterator {
         self.0
     }
 }
 
-impl Protected<*mut _CassIterator> for UserTypeIterator {
+impl Protected<*mut _CassIterator> for UserTypeIterator<'_> {
     fn build(inner: *mut _CassIterator) -> Self {
         if inner.is_null() {
             panic!("Unexpected null pointer")
         };
-        UserTypeIterator(inner)
+        UserTypeIterator(inner, PhantomData)
     }
 }
 
-impl ProtectedInner<*mut _CassIterator> for AggregateIterator {
+impl ProtectedInner<*mut _CassIterator> for AggregateIterator<'_> {
     fn inner(&self) -> *mut _CassIterator {
         self.0
     }
 }
 
-impl Protected<*mut _CassIterator> for AggregateIterator {
+impl Protected<*mut _CassIterator> for AggregateIterator<'_> {
     fn build(inner: *mut _CassIterator) -> Self {
         if inner.is_null() {
             panic!("Unexpected null pointer")
         };
-        AggregateIterator(inner)
+        AggregateIterator(inner, PhantomData)
     }
 }
 
-impl ProtectedInner<*mut _CassIterator> for FunctionIterator {
+impl ProtectedInner<*mut _CassIterator> for FunctionIterator<'_> {
     fn inner(&self) -> *mut _CassIterator {
         self.0
     }
 }
 
-impl Protected<*mut _CassIterator> for FunctionIterator {
+impl Protected<*mut _CassIterator> for FunctionIterator<'_> {
     fn build(inner: *mut _CassIterator) -> Self {
         if inner.is_null() {
             panic!("Unexpected null pointer")
         };
-        FunctionIterator(inner)
+        FunctionIterator(inner, PhantomData)
     }
 }
 
-impl ProtectedInner<*mut _CassIterator> for KeyspaceIterator {
+impl ProtectedInner<*mut _CassIterator> for KeyspaceIterator<'_> {
     fn inner(&self) -> *mut _CassIterator {
         self.0
     }
 }
 
-impl Protected<*mut _CassIterator> for KeyspaceIterator {
+impl Protected<*mut _CassIterator> for KeyspaceIterator<'_> {
     fn build(inner: *mut _CassIterator) -> Self {
         if inner.is_null() {
             panic!("Unexpected null pointer")
         };
-        KeyspaceIterator(inner)
+        KeyspaceIterator(inner, PhantomData)
     }
 }
 
-impl ProtectedInner<*mut _CassIterator> for FieldIterator {
+impl ProtectedInner<*mut _CassIterator> for FieldIterator<'_> {
     fn inner(&self) -> *mut _CassIterator {
         self.0
     }
 }
-impl Protected<*mut _CassIterator> for FieldIterator {
+impl Protected<*mut _CassIterator> for FieldIterator<'_> {
     fn build(inner: *mut _CassIterator) -> Self {
         if inner.is_null() {
             panic!("Unexpected null pointer")
         };
-        FieldIterator(inner)
+        FieldIterator(inner, PhantomData)
     }
 }
 
-impl ProtectedInner<*mut _CassIterator> for ColumnIterator {
+impl ProtectedInner<*mut _CassIterator> for ColumnIterator<'_> {
     fn inner(&self) -> *mut _CassIterator {
         self.0
     }
 }
 
-impl Protected<*mut _CassIterator> for ColumnIterator {
+impl Protected<*mut _CassIterator> for ColumnIterator<'_> {
     fn build(inner: *mut _CassIterator) -> Self {
         if inner.is_null() {
             panic!("Unexpected null pointer")
         };
-        ColumnIterator(inner)
+        ColumnIterator(inner, PhantomData)
     }
 }
 
-impl ProtectedInner<*mut _CassIterator> for TableIterator {
+impl ProtectedInner<*mut _CassIterator> for TableIterator<'_> {
     fn inner(&self) -> *mut _CassIterator {
         self.0
     }
 }
 
-impl Protected<*mut _CassIterator> for TableIterator {
+impl Protected<*mut _CassIterator> for TableIterator<'_> {
     fn build(inner: *mut _CassIterator) -> Self {
         if inner.is_null() {
             panic!("Unexpected null pointer")
         };
-        TableIterator(inner)
+        TableIterator(inner, PhantomData)
     }
 }
 
-impl ProtectedInner<*mut _CassIterator> for MapIterator {
+impl ProtectedInner<*mut _CassIterator> for MapIterator<'_> {
     fn inner(&self) -> *mut _CassIterator {
         self.0
     }
 }
 
-impl Protected<*mut _CassIterator> for MapIterator {
+impl Protected<*mut _CassIterator> for MapIterator<'_> {
     fn build(inner: *mut _CassIterator) -> Self {
         if inner.is_null() {
             panic!("Unexpected null pointer")
         };
-        MapIterator(inner)
+        MapIterator(inner, PhantomData)
     }
 }
 
-impl ProtectedInner<*mut _CassIterator> for SetIterator {
+impl ProtectedInner<*mut _CassIterator> for SetIterator<'_> {
     fn inner(&self) -> *mut _CassIterator {
         self.0
     }
 }
 
-impl Protected<*mut _CassIterator> for SetIterator {
+impl Protected<*mut _CassIterator> for SetIterator<'_> {
     fn build(inner: *mut _CassIterator) -> Self {
         if inner.is_null() {
             panic!("Unexpected null pointer")
         };
-        SetIterator(inner)
+        SetIterator(inner, PhantomData)
     }
 }
 
-/// Iterater over the set's metadata entries(??)
+/// Iterator over the values in a set.
+///
+/// A Cassandra iterator is a `LendingIterator` because it borrows from some
+/// underlying value, but owns a single item. Each time `next()` is invoked it
+/// decodes the current item into that item, thus invalidating its previous
+/// value.
 #[derive(Debug)]
-pub struct SetIterator(*mut _CassIterator);
-
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
-unsafe impl Send for SetIterator {}
-
-// impl<'a> Display for &'a SetIterator {
-//    fn fmt(&self, f:&mut Formatter) -> fmt::Result {
-//        for item in self {
-//            try!(write!(f, "{}\t", item));
-//        }
-//        Ok(())
-//    }
-// }
-
-impl Drop for SetIterator {
+pub struct SetIterator<'a>(*mut _CassIterator, PhantomData<&'a _CassValue>);
+
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
+unsafe impl Send for SetIterator<'_> {}
+unsafe impl Sync for SetIterator<'_> {}
+
+impl Drop for SetIterator<'_> {
     fn drop(&mut self) {
         unsafe { cass_iterator_free(self.0) }
     }
 }
 
-impl Iterator for SetIterator {
-    type Item = Value;
+impl LendingIterator for SetIterator<'_> {
+    type Item<'a> = Value<'a> where Self: 'a;
 
-    fn next(&mut self) -> Option<<Self as Iterator>::Item> {
+    fn next(&mut self) -> Option<<Self as LendingIterator>::Item<'_>> {
         unsafe {
             match cass_iterator_next(self.0) {
                 cass_false => None,
@@ -402,75 +477,49 @@ impl Iterator for SetIterator {
     }
 }
 
-impl SetIterator {
-    fn get_value(&mut self) -> Value {
+impl SetIterator<'_> {
+    fn get_value(&self) -> Value {
         unsafe { Value::build(cass_iterator_get_value(self.0)) }
     }
 }
 
-/// An iterator over the k/v pair in the map
+/// An iterator over the k/v pairs in a map.
 #[derive(Debug)]
-pub struct MapIterator(*mut _CassIterator);
-
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
-unsafe impl Send for MapIterator {}
-
-impl MapIterator {
-    fn get_key(&mut self) -> Value {
+///
+/// A Cassandra iterator is a `LendingIterator` because it borrows from some
+/// underlying value, but owns a single item. Each time `next()` is invoked it
+/// decodes the current item into that item, thus invalidating its previous
+/// value.
+pub struct MapIterator<'a>(*mut _CassIterator, PhantomData<&'a _CassValue>);
+
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
+unsafe impl Send for MapIterator<'_> {}
+unsafe impl Sync for MapIterator<'_> {}
+
+impl MapIterator<'_> {
+    fn get_key(&self) -> Value {
         unsafe { Value::build(cass_iterator_get_map_key(self.0)) }
     }
-    fn get_value(&mut self) -> Value {
+    fn get_value(&self) -> Value {
         unsafe { Value::build(cass_iterator_get_map_value(self.0)) }
     }
 
     /// Gets the next k/v pair in the map
-    pub fn get_pair(&mut self) -> (Value, Value) {
+    pub fn get_pair(&self) -> (Value, Value) {
         (self.get_key(), self.get_value())
     }
 }
 
-/// An iterator over the elements of a Cassandra tuple
-#[derive(Debug)]
-pub struct TupleIterator(pub *mut _CassIterator);
-
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
-unsafe impl Send for TupleIterator {}
-
-impl Drop for TupleIterator {
-    fn drop(&mut self) {
-        unsafe { cass_iterator_free(self.0) }
-    }
-}
-
-impl Iterator for TupleIterator {
-    type Item = Value;
-    fn next(&mut self) -> Option<<Self as Iterator>::Item> {
-        unsafe {
-            match cass_iterator_next(self.0) {
-                cass_false => None,
-                cass_true => Some(self.get_value()),
-            }
-        }
-    }
-}
-
-impl TupleIterator {
-    fn get_value(&mut self) -> Value {
-        unsafe { Value::build(cass_iterator_get_value(self.0)) }
-    }
-}
-
-impl Drop for MapIterator {
+impl Drop for MapIterator<'_> {
     fn drop(&mut self) {
         unsafe { cass_iterator_free(self.0) }
     }
 }
 
-impl Iterator for MapIterator {
-    type Item = (Value, Value);
-    fn next(&mut self) -> Option<<Self as Iterator>::Item> {
+impl LendingIterator for MapIterator<'_> {
+    type Item<'a> = (Value<'a>, Value<'a>) where Self: 'a;
+    fn next(&mut self) -> Option<<Self as LendingIterator>::Item<'_>> {
         unsafe {
             match cass_iterator_next(self.0) {
                 cass_false => None,

@@ -4,12 +4,12 @@ use crate::cassandra_sys::CassLogMessage;
 use crate::cassandra::util::ProtectedInner;
 use crate::cassandra_sys::cass_log_set_callback;
 use crate::cassandra_sys::cass_log_set_level;
-use crate::cassandra_sys::CassLogCallback;
+
 // use cassandra_sys::cass_log_set_queue_size; @deprecated
 
 #[cfg(feature = "slog")]
 use slog;
-use std::borrow::Borrow;
+
 use std::boxed::Box;
 use std::ffi::CStr;
 use std::os::raw;

@@ -1,4 +1,3 @@
-use crate::cassandra::util::Protected;
 use crate::cassandra_sys::CassMetrics as _CassMetrics;
 
 /// Metrics about the current session.

@@ -10,9 +10,10 @@ use crate::cassandra_sys::CassRetryPolicy as _RetryPolicy;
 #[derive(Debug)]
 pub struct RetryPolicy(*mut _RetryPolicy);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
 unsafe impl Send for RetryPolicy {}
+unsafe impl Sync for RetryPolicy {}
 
 impl ProtectedInner<*mut _RetryPolicy> for RetryPolicy {
     fn inner(&self) -> *mut _RetryPolicy {

@@ -10,7 +10,7 @@ use crate::cassandra_sys::cass_prepared_parameter_data_type_by_name_n;
 use crate::cassandra_sys::cass_prepared_parameter_name;
 use crate::cassandra_sys::CassPrepared as _PreparedStatement;
 use std::os::raw::c_char;
-use std::{mem, slice, str};
+use std::{slice, str};
 
 /// A statement that has been prepared against at least one Cassandra node.
 /// Instances of this class should not be created directly, but through Session.prepare().
@@ -78,16 +78,14 @@ impl PreparedStatement {
 
     /// Gets the data type of a parameter at the specified index.
     ///
-    /// Returns a reference to the data type of the parameter. Do not free
-    /// this reference as it is bound to the lifetime of the prepared.
+    /// Returns a reference to the data type of the parameter.
     pub fn parameter_data_type(&self, index: usize) -> ConstDataType {
         unsafe { ConstDataType::build(cass_prepared_parameter_data_type(self.0, index)) }
     }
 
     /// Gets the data type of a parameter for the specified name.
     ///
-    /// Returns a reference to the data type of the parameter. Do not free
-    /// this reference as it is bound to the lifetime of the prepared.
+    /// Returns a reference to the data type of the parameter.
     pub fn parameter_data_type_by_name(&self, name: &str) -> ConstDataType {
         unsafe {
             let name_ptr = name.as_ptr() as *const c_char;

@@ -4,6 +4,7 @@
 
 use crate::cassandra::data_type::ConstDataType;
 use crate::cassandra::error::*;
+use crate::cassandra::iterator::LendingIterator;
 use crate::cassandra::row::Row;
 use crate::cassandra::util::{Protected, ProtectedInner};
 use crate::cassandra::value::ValueType;
@@ -26,13 +27,12 @@ use crate::cassandra_sys::cass_true;
 use crate::cassandra_sys::CassIterator as _CassIterator;
 use crate::cassandra_sys::CassResult as _CassResult;
 
-use std::ffi::CString;
 use std::fmt;
 use std::fmt::Debug;
 use std::fmt::Display;
 use std::fmt::Formatter;
 use std::marker::PhantomData;
-use std::mem;
+
 use std::slice;
 use std::str;
 
@@ -61,7 +61,8 @@ impl Protected<*const _CassResult> for CassResult {
 impl Debug for CassResult {
     fn fmt(&self, f: &mut Formatter) -> fmt::Result {
         writeln!(f, "Result row count: {:?}", self.row_count())?;
-        for row in self.iter() {
+        let mut iter = self.iter();
+        while let Some(row) = iter.next() {
             writeln!(f, "{:?}", row)?;
         }
         Ok(())
@@ -71,7 +72,8 @@ impl Debug for CassResult {
 impl Display for CassResult {
     fn fmt(&self, f: &mut Formatter) -> fmt::Result {
         writeln!(f, "Result row count: {}", self.row_count())?;
-        for row in self.iter() {
+        let mut iter = self.iter();
+        while let Some(row) = iter.next() {
             writeln!(f, "{}", row)?;
         }
         Ok(())
@@ -171,24 +173,44 @@ impl CassResult {
     }
 }
 
-/// An iterator over the results of a query.
-/// The result holds the data, so it must last for at least the lifetime of the iterator.
+/// An iterator over the results of a query. The result holds the data, so
+/// the result must last for at least the lifetime of the iterator.
+///
+/// This is a lending iterator (you must stop using each item before you move to
+/// the next), and so it does not implement `std::iter::Iterator`. The best way
+/// to use it is as follows:
+///
+/// ```no_run
+/// # use cassandra_cpp::*;
+/// # let result: CassResult = unimplemented!();
+/// let mut iter = result.iter();
+/// while let Some(row) = iter.next() {
+///   // ... do something with `row` ...
+/// }
+/// ```
+///
+/// A Cassandra iterator is a `LendingIterator` because it borrows from some
+/// underlying value, but owns a single item. Each time `next()` is invoked it
+/// decodes the current item into that item, thus invalidating its previous
+/// value.
 #[derive(Debug)]
-pub struct ResultIterator<'a>(pub *mut _CassIterator, usize, PhantomData<&'a CassResult>);
+pub struct ResultIterator<'a>(*mut _CassIterator, usize, PhantomData<&'a _CassResult>);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
-unsafe impl<'a> Send for ResultIterator<'a> {}
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
+unsafe impl Send for ResultIterator<'_> {}
+unsafe impl Sync for ResultIterator<'_> {}
 
 impl<'a> Drop for ResultIterator<'a> {
     fn drop(&mut self) {
         unsafe { cass_iterator_free(self.0) }
     }
 }
 
-impl<'a> Iterator for ResultIterator<'a> {
-    type Item = Row<'a>;
-    fn next(&mut self) -> Option<<Self as Iterator>::Item> {
+impl LendingIterator for ResultIterator<'_> {
+    type Item<'a> = Row<'a> where Self: 'a;
+
+    fn next(&mut self) -> Option<<Self as LendingIterator>::Item<'_>> {
         unsafe {
             match cass_iterator_next(self.0) {
                 cass_false => None,
@@ -202,18 +224,9 @@ impl<'a> Iterator for ResultIterator<'a> {
     }
 }
 
-impl<'a> ResultIterator<'a> {
-    /// Gets the next row in the result set
-    pub fn get_row(&mut self) -> Row<'a> {
+impl ResultIterator<'_> {
+    /// Gets the current row in the result set
+    pub fn get_row(&self) -> Row {
         unsafe { Row::build(cass_iterator_get_row(self.0)) }
     }
 }
-
-impl<'a> IntoIterator for &'a CassResult {
-    type Item = Row<'a>;
-    type IntoIter = ResultIterator<'a>;
-
-    fn into_iter(self) -> Self::IntoIter {
-        self.iter()
-    }
-}

@@ -4,7 +4,7 @@ use crate::cassandra::error::*;
 
 use crate::cassandra::inet::Inet;
 use crate::cassandra::iterator::{MapIterator, SetIterator, UserTypeIterator};
-use crate::cassandra::result::CassResult;
+
 use crate::cassandra::util::{Protected, ProtectedInner};
 use crate::cassandra::uuid::Uuid;
 use crate::cassandra::value::Value;
@@ -18,28 +18,29 @@ use crate::cassandra_sys::cass_row_get_column_by_name_n;
 use crate::cassandra_sys::cass_true;
 use crate::cassandra_sys::CassIterator as _CassIterator;
 use crate::cassandra_sys::CassRow as _Row;
+use crate::LendingIterator;
 use std::fmt;
 use std::fmt::Debug;
 use std::fmt::Display;
 use std::fmt::Formatter;
-use std::iter;
-use std::iter::IntoIterator;
 use std::marker::PhantomData;
 use std::os::raw::c_char;
 
 /// A collection of column values. Read-only, so thread-safe.
-pub struct Row<'a>(*const _Row, PhantomData<&'a CassResult>);
+//
+// Borrowed immutably.
+pub struct Row<'a>(*const _Row, PhantomData<&'a _Row>);
 
-unsafe impl<'a> Sync for Row<'a> {}
-unsafe impl<'a> Send for Row<'a> {}
+unsafe impl Sync for Row<'_> {}
+unsafe impl Send for Row<'_> {}
 
-impl<'a> ProtectedInner<*const _Row> for Row<'a> {
+impl ProtectedInner<*const _Row> for Row<'_> {
     fn inner(&self) -> *const _Row {
         self.0
     }
 }
 
-impl<'a> Protected<*const _Row> for Row<'a> {
+impl Protected<*const _Row> for Row<'_> {
     fn build(inner: *const _Row) -> Self {
         if inner.is_null() {
             panic!("Unexpected null pointer")
@@ -48,30 +49,32 @@ impl<'a> Protected<*const _Row> for Row<'a> {
     }
 }
 
-impl<'a> Debug for Row<'a> {
+impl Debug for Row<'_> {
     fn fmt(&self, f: &mut Formatter) -> fmt::Result {
-        for column in self {
+        let mut iter = self.iter();
+        while let Some(column) = iter.next() {
             write!(f, "{:?}\t", Value::build(column.inner()))?;
         }
         Ok(())
     }
 }
 
-impl<'a> Display for Row<'a> {
+impl Display for Row<'_> {
     fn fmt(&self, f: &mut Formatter) -> fmt::Result {
-        for column in self {
+        let mut iter = self.iter();
+        while let Some(column) = iter.next() {
             write!(f, "{}\t", Value::build(column.inner()))?;
         }
         Ok(())
     }
 }
 
-/// Auto inferencing conversion from c* to rust
+/// Auto inferencing conversion from Cassandra to Rust.
 pub trait AsRustType<T> {
-    /// convert while reading cassandra columns
+    /// Convert Cassandra column by index.
     fn get(&self, index: usize) -> Result<T>;
 
-    /// convert while reading cassandra columns by name
+    /// Convert Cassandra column by name.
     fn get_by_name<S>(&self, name: S) -> Result<T>
     where
         S: Into<String>;
@@ -226,13 +229,16 @@ impl AsRustType<Inet> for Row<'_> {
     }
 }
 
-impl AsRustType<SetIterator> for Row<'_> {
-    fn get(&self, index: usize) -> Result<SetIterator> {
+impl<'a> AsRustType<SetIterator<'a>> for Row<'a> {
+    // The iterator is newly-created here, but it borrows the data it iterates
+    // over from the row (i.e., from its underlying result). Thus its lifetime
+    // parameter is the same as the row's.
+    fn get(&self, index: usize) -> Result<SetIterator<'a>> {
         let col = self.get_column(index)?;
         col.get_set()
     }
 
-    fn get_by_name<S>(&self, name: S) -> Result<SetIterator>
+    fn get_by_name<S>(&self, name: S) -> Result<SetIterator<'a>>
     where
         S: Into<String>,
     {
@@ -241,13 +247,16 @@ impl AsRustType<SetIterator> for Row<'_> {
     }
 }
 
-impl AsRustType<MapIterator> for Row<'_> {
-    fn get(&self, index: usize) -> Result<MapIterator> {
+impl<'a> AsRustType<MapIterator<'a>> for Row<'a> {
+    // The iterator is newly-created here, but it borrows the data it iterates
+    // over from the row (i.e., from its underlying result). Thus its lifetime
+    // parameter is the same as the row's.
+    fn get(&self, index: usize) -> Result<MapIterator<'a>> {
         let col = self.get_column(index)?;
         col.get_map()
     }
 
-    fn get_by_name<S>(&self, name: S) -> Result<MapIterator>
+    fn get_by_name<S>(&self, name: S) -> Result<MapIterator<'a>>
     where
         S: Into<String>,
     {
@@ -256,13 +265,16 @@ impl AsRustType<MapIterator> for Row<'_> {
     }
 }
 
-impl AsRustType<UserTypeIterator> for Row<'_> {
-    fn get(&self, index: usize) -> Result<UserTypeIterator> {
+impl<'a> AsRustType<UserTypeIterator<'a>> for Row<'a> {
+    // The iterator is newly-created here, but it borrows the data it iterates
+    // over from the row (i.e., from its underlying result). Thus its lifetime
+    // parameter is the same as the row's.
+    fn get(&self, index: usize) -> Result<UserTypeIterator<'a>> {
         let col = self.get_column(index)?;
         col.get_user_type()
     }
 
-    fn get_by_name<S>(&self, name: S) -> Result<UserTypeIterator>
+    fn get_by_name<S>(&self, name: S) -> Result<UserTypeIterator<'a>>
     where
         S: Into<String>,
     {
@@ -333,7 +345,7 @@ impl AsRustType<BigDecimal> for Row<'_> {
 
 impl<'a> Row<'a> {
     /// Get a particular column by index
-    pub fn get_column(&self, index: usize) -> Result<Value> {
+    pub fn get_column(&self, index: usize) -> Result<Value<'a>> {
         unsafe {
             let col = cass_row_get_column(self.0, index);
             if col.is_null() {
@@ -345,7 +357,7 @@ impl<'a> Row<'a> {
     }
 
     /// Get a particular column by name
-    pub fn get_column_by_name<S>(&self, name: S) -> Result<Value>
+    pub fn get_column_by_name<S>(&self, name: S) -> Result<Value<'a>>
     where
         S: Into<String>,
     {
@@ -360,39 +372,38 @@ impl<'a> Row<'a> {
             }
         }
     }
+
+    /// Creates a new iterator for the specified row. This can be
+    /// used to iterate over columns in a row.
+    fn iter(&'a self) -> RowIterator<'a> {
+        unsafe { RowIterator(cass_iterator_from_row(self.0), PhantomData) }
+    }
 }
 
 /// An iterator over the columns in a row
+///
+/// A Cassandra iterator is a `LendingIterator` because it borrows from some
+/// underlying value, but owns a single item. Each time `next()` is invoked it
+/// decodes the current item into that item, thus invalidating its previous
+/// value.
 #[derive(Debug)]
-pub struct RowIterator(pub *mut _CassIterator);
+pub struct RowIterator<'a>(*mut _CassIterator, PhantomData<&'a _Row>);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
-unsafe impl Send for RowIterator {}
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
+unsafe impl Send for RowIterator<'_> {}
+unsafe impl Sync for RowIterator<'_> {}
 
-impl Drop for RowIterator {
+impl Drop for RowIterator<'_> {
     fn drop(&mut self) {
         unsafe { cass_iterator_free(self.0) }
     }
 }
 
-impl iter::Iterator for RowIterator {
-    type Item = Value;
-
-    fn next(&mut self) -> Option<<Self as Iterator>::Item> {
-        unsafe {
-            match cass_iterator_next(self.0) {
-                cass_false => None,
-                cass_true => Some(Value::build(cass_iterator_get_column(self.0))),
-            }
-        }
-    }
-}
-
-impl<'a> Iterator for &'a RowIterator {
-    type Item = Value;
+impl LendingIterator for RowIterator<'_> {
+    type Item<'a> = Value<'a> where Self: 'a;
 
-    fn next(&mut self) -> Option<<Self as Iterator>::Item> {
+    fn next(&mut self) -> Option<<Self as LendingIterator>::Item<'_>> {
         unsafe {
             match cass_iterator_next(self.0) {
                 cass_false => None,
@@ -401,31 +412,3 @@ impl<'a> Iterator for &'a RowIterator {
         }
     }
 }
-
-impl Display for RowIterator {
-    fn fmt(&self, f: &mut Formatter) -> fmt::Result {
-        for item in self {
-            write!(f, "{}\t", Value::build(item.inner()))?;
-        }
-        Ok(())
-    }
-}
-
-impl IntoIterator for Row<'_> {
-    type Item = Value;
-    type IntoIter = RowIterator;
-
-    /// Creates a new iterator for the specified row. This can be
-    /// used to iterate over columns in a row.
-    fn into_iter(self) -> Self::IntoIter {
-        unsafe { RowIterator(cass_iterator_from_row(self.0)) }
-    }
-}
-
-impl<'a> IntoIterator for &'a Row<'_> {
-    type Item = Value;
-    type IntoIter = RowIterator;
-    fn into_iter(self) -> Self::IntoIter {
-        unsafe { RowIterator(cass_iterator_from_row(self.0)) }
-    }
-}

@@ -18,31 +18,37 @@ use crate::cassandra_sys::cass_aggregate_meta_state_type;
 use crate::cassandra_sys::cass_iterator_fields_from_aggregate_meta;
 use crate::cassandra_sys::raw2utf8;
 use crate::cassandra_sys::CassAggregateMeta as _CassAggregateMeta;
-use std::mem;
+use std::marker::PhantomData;
+
 use std::os::raw::c_char;
 
-/// Metadata about a cassandra aggregate
+/// Metadata about a Cassandra aggregate
+//
+// Borrowed immutably.
 #[derive(Debug)]
-pub struct AggregateMeta(*const _CassAggregateMeta);
+pub struct AggregateMeta<'a>(
+    *const _CassAggregateMeta,
+    PhantomData<&'a _CassAggregateMeta>,
+);
 
-impl ProtectedInner<*const _CassAggregateMeta> for AggregateMeta {
+impl ProtectedInner<*const _CassAggregateMeta> for AggregateMeta<'_> {
     fn inner(&self) -> *const _CassAggregateMeta {
         self.0
     }
 }
 
-impl Protected<*const _CassAggregateMeta> for AggregateMeta {
+impl Protected<*const _CassAggregateMeta> for AggregateMeta<'_> {
     fn build(inner: *const _CassAggregateMeta) -> Self {
         if inner.is_null() {
             panic!("Unexpected null pointer")
         };
-        AggregateMeta(inner)
+        AggregateMeta(inner, PhantomData)
     }
 }
 
-impl AggregateMeta {
+impl<'a> AggregateMeta<'a> {
     /// An iterator over the fields of an aggregate
-    pub fn fields_iter(&self) -> FieldIterator {
+    pub fn fields_iter(&self) -> FieldIterator<'a> {
         unsafe { FieldIterator::build(cass_iterator_fields_from_aggregate_meta(self.0)) }
     }
 
@@ -72,39 +78,39 @@ impl AggregateMeta {
     }
 
     /// Gets the aggregate's argument type for the provided index.
-    pub fn argument_type(&self, index: usize) -> ConstDataType {
+    pub fn argument_type(&self, index: usize) -> ConstDataType<'a> {
         // TODO: can return NULL
         unsafe { ConstDataType::build(cass_aggregate_meta_argument_type(self.0, index)) }
     }
 
     /// Gets the aggregate's argument return type.
-    pub fn return_type(&self) -> ConstDataType {
+    pub fn return_type(&self) -> ConstDataType<'a> {
         unsafe { ConstDataType::build(cass_aggregate_meta_return_type(self.0)) }
     }
 
     /// Gets the aggregate's argument state type.
-    pub fn state_type(&self) -> ConstDataType {
+    pub fn state_type(&self) -> ConstDataType<'a> {
         unsafe { ConstDataType::build(cass_aggregate_meta_state_type(self.0)) }
     }
 
     /// Gets the function metadata for the aggregate's state function.
-    pub fn state_func(&self) -> FunctionMeta {
+    pub fn state_func(&self) -> FunctionMeta<'a> {
         unsafe { FunctionMeta::build(cass_aggregate_meta_state_func(self.0)) }
     }
 
     /// Gets the function metadata for the aggregates's final function.
-    pub fn final_func(&self) -> FunctionMeta {
+    pub fn final_func(&self) -> FunctionMeta<'a> {
         unsafe { FunctionMeta::build(cass_aggregate_meta_final_func(self.0)) }
     }
 
     ///  Gets the initial condition value for the aggregate.
-    pub fn init_cond(&self) -> Value {
+    pub fn init_cond(&self) -> Value<'a> {
         unsafe { Value::build(cass_aggregate_meta_init_cond(self.0)) }
     }
 
     ///  Gets a metadata field for the provided name. Metadata fields allow direct
     /// access to the column data found in the underlying "aggregates" metadata table.
-    pub fn field_by_name(&self, name: &str) -> Option<Value> {
+    pub fn field_by_name(&self, name: &str) -> Option<Value<'a>> {
         unsafe {
             let name_ptr = name.as_ptr() as *const c_char;
             let agg = cass_aggregate_meta_field_by_name_n(self.0, name_ptr, name.len());

@@ -11,33 +11,35 @@ use crate::cassandra_sys::cass_iterator_fields_from_column_meta;
 use crate::cassandra_sys::CassColumnMeta as _CassColumnMeta;
 use crate::cassandra_sys::CassColumnType as _CassColumnType;
 
-/// Column metadata
-#[derive(Debug)]
-pub struct ColumnMeta(*const _CassColumnMeta);
-
-use std::mem;
+use std::marker::PhantomData;
 use std::os::raw::c_char;
 use std::slice;
 use std::str;
 
-impl ProtectedInner<*const _CassColumnMeta> for ColumnMeta {
+/// Column metadata
+//
+// Borrowed immutably.
+#[derive(Debug)]
+pub struct ColumnMeta<'a>(*const _CassColumnMeta, PhantomData<&'a _CassColumnMeta>);
+
+impl ProtectedInner<*const _CassColumnMeta> for ColumnMeta<'_> {
     fn inner(&self) -> *const _CassColumnMeta {
         self.0
     }
 }
 
-impl Protected<*const _CassColumnMeta> for ColumnMeta {
+impl Protected<*const _CassColumnMeta> for ColumnMeta<'_> {
     fn build(inner: *const _CassColumnMeta) -> Self {
         if inner.is_null() {
             panic!("Unexpected null pointer")
         };
-        ColumnMeta(inner)
+        ColumnMeta(inner, PhantomData)
     }
 }
 
-impl ColumnMeta {
+impl<'a> ColumnMeta<'a> {
     /// returns an iterator over the fields of this column
-    pub fn field_iter(&mut self) -> FieldIterator {
+    pub fn field_iter(&self) -> FieldIterator<'a> {
         unsafe { FieldIterator::build(cass_iterator_fields_from_column_meta(self.0)) }
     }
 
@@ -58,13 +60,13 @@ impl ColumnMeta {
     }
 
     /// Gets the data type of the column.
-    pub fn data_type(&self) -> ConstDataType {
+    pub fn data_type(&self) -> ConstDataType<'a> {
         unsafe { ConstDataType::build(cass_column_meta_data_type(self.0)) }
     }
 
     /// Gets a metadata field for the provided name. Metadata fields allow direct
     /// access to the column data found in the underlying "columns" metadata table.
-    pub fn field_by_name(&self, name: &str) -> Option<Value> {
+    pub fn field_by_name(&self, name: &str) -> Option<Value<'a>> {
         unsafe {
             let name_ptr = name.as_ptr() as *const c_char;
             let field = cass_column_meta_field_by_name_n(self.0, name_ptr, name.len());

@@ -17,33 +17,35 @@ use crate::cassandra_sys::cass_function_meta_return_type;
 use crate::cassandra_sys::cass_iterator_fields_from_function_meta;
 use crate::cassandra_sys::cass_true;
 use crate::cassandra_sys::CassFunctionMeta as _CassFunctionMeta;
-use crate::cassandra_sys::CASS_OK;
 
+use std::marker::PhantomData;
 use std::os::raw::c_char;
-use std::{mem, slice, str};
+use std::{slice, str};
 
 /// The metadata for a function
+//
+// Borrowed immutably.
 #[derive(Debug)]
-pub struct FunctionMeta(*const _CassFunctionMeta);
+pub struct FunctionMeta<'a>(*const _CassFunctionMeta, PhantomData<&'a _CassFunctionMeta>);
 
-impl ProtectedInner<*const _CassFunctionMeta> for FunctionMeta {
+impl ProtectedInner<*const _CassFunctionMeta> for FunctionMeta<'_> {
     fn inner(&self) -> *const _CassFunctionMeta {
         self.0
     }
 }
 
-impl Protected<*const _CassFunctionMeta> for FunctionMeta {
+impl Protected<*const _CassFunctionMeta> for FunctionMeta<'_> {
     fn build(inner: *const _CassFunctionMeta) -> Self {
         if inner.is_null() {
             panic!("Unexpected null pointer")
         };
-        FunctionMeta(inner)
+        FunctionMeta(inner, PhantomData)
     }
 }
 
-impl FunctionMeta {
-    /// Iterator over the fields in this function
-    pub fn fields_iter(&self) -> FieldIterator {
+impl<'a> FunctionMeta<'a> {
+    /// Iterator over the fields in this function.
+    pub fn fields_iter(&self) -> FieldIterator<'a> {
         unsafe { FieldIterator::build(cass_iterator_fields_from_function_meta(self.0)) }
     }
 
@@ -108,18 +110,23 @@ impl FunctionMeta {
     }
 
     /// Gets the function's argument name and type for the provided index.
-    pub fn argument(&self, index: usize) -> Result<()> {
+    pub fn argument(&self, index: usize) -> Result<(String, ConstDataType<'a>)> {
         let mut name = std::ptr::null();
         let mut name_length = 0;
         let mut data_type = std::ptr::null();
         unsafe {
             cass_function_meta_argument(self.0, index, &mut name, &mut name_length, &mut data_type)
-                .to_result(())
+                .to_result(())?;
+            let name = str::from_utf8(slice::from_raw_parts(name as *const u8, name_length))
+                .expect("must be utf8")
+                .to_owned();
+            let data_type = ConstDataType::build(data_type);
+            Ok((name, data_type))
         }
     }
 
     /// Gets the function's argument and type for the provided name.
-    pub fn argument_type_by_name(&self, name: &str) -> ConstDataType {
+    pub fn argument_type_by_name(&self, name: &str) -> ConstDataType<'a> {
         unsafe {
             let name_ptr = name.as_ptr() as *const c_char;
             // TODO: can return NULL
@@ -132,13 +139,13 @@ impl FunctionMeta {
     }
 
     /// Gets the return type of the function.
-    pub fn return_type(&self) -> ConstDataType {
+    pub fn return_type(&self) -> ConstDataType<'a> {
         unsafe { ConstDataType::build(cass_function_meta_return_type(self.0)) }
     }
 
     /// Gets a metadata field for the provided name. Metadata fields allow direct
     /// access to the column data found in the underlying "functions" metadata table.
-    pub fn field_by_name(&self, name: &str) -> Value {
+    pub fn field_by_name(&self, name: &str) -> Value<'a> {
         unsafe {
             let name_ptr = name.as_ptr() as *const c_char;
             // TODO: can return NULL

@@ -25,44 +25,48 @@ use crate::cassandra_sys::cass_keyspace_meta_name;
 use crate::cassandra_sys::cass_keyspace_meta_table_by_name_n;
 use crate::cassandra_sys::cass_keyspace_meta_user_type_by_name_n;
 use crate::cassandra_sys::raw2utf8;
-use std::mem;
+
+use std::marker::PhantomData;
+
 use std::os::raw::c_char;
 
 /// A snapshot of the schema's metadata.
+//
+// Borrowed immutably.
 #[derive(Debug)]
-pub struct KeyspaceMeta(*const _CassKeyspaceMeta);
+pub struct KeyspaceMeta<'a>(*const _CassKeyspaceMeta, PhantomData<&'a _CassKeyspaceMeta>);
 
-impl ProtectedInner<*const _CassKeyspaceMeta> for KeyspaceMeta {
+impl ProtectedInner<*const _CassKeyspaceMeta> for KeyspaceMeta<'_> {
     fn inner(&self) -> *const _CassKeyspaceMeta {
         self.0
     }
 }
 
-impl Protected<*const _CassKeyspaceMeta> for KeyspaceMeta {
+impl Protected<*const _CassKeyspaceMeta> for KeyspaceMeta<'_> {
     fn build(inner: *const _CassKeyspaceMeta) -> Self {
         if inner.is_null() {
             panic!("Unexpected null pointer")
         };
-        KeyspaceMeta(inner)
+        KeyspaceMeta(inner, PhantomData)
     }
 }
 
 #[derive(Debug)]
-pub struct MetadataFieldValue(*const _CassValue);
+pub struct MetadataFieldValue<'a>(*const _CassValue, PhantomData<&'a _CassValue>);
 
-impl KeyspaceMeta {
+impl<'a> KeyspaceMeta<'a> {
     /// Iterator over the aggregates in this keyspace
-    pub fn aggregrates_iter(&self) -> AggregateIterator {
+    pub fn aggregrates_iter(&self) -> AggregateIterator<'a> {
         unsafe { AggregateIterator::build(cass_iterator_aggregates_from_keyspace_meta(self.0)) }
     }
 
     /// Iterator over the field in this keyspace
-    pub fn fields_iter(&self) -> FieldIterator {
+    pub fn fields_iter(&self) -> FieldIterator<'a> {
         unsafe { FieldIterator::build(cass_iterator_fields_from_keyspace_meta(self.0)) }
     }
 
     /// Gets the table metadata for the provided table name.
-    pub fn table_by_name(&self, name: &str) -> Option<TableMeta> {
+    pub fn table_by_name(&self, name: &str) -> Option<TableMeta<'a>> {
         unsafe {
             let name_ptr = name.as_ptr() as *const c_char;
             let value = cass_keyspace_meta_table_by_name_n(self.0, name_ptr, name.len());
@@ -75,7 +79,7 @@ impl KeyspaceMeta {
     }
 
     /// Gets the data type for the provided type name.
-    pub fn user_type_by_name(&self, name: &str) -> Option<ConstDataType> {
+    pub fn user_type_by_name(&self, name: &str) -> Option<ConstDataType<'a>> {
         unsafe {
             let name_ptr = name.as_ptr() as *const c_char;
             let value = cass_keyspace_meta_user_type_by_name_n(self.0, name_ptr, name.len());
@@ -88,7 +92,11 @@ impl KeyspaceMeta {
     }
 
     /// Gets the function metadata for the provided function name.
-    pub fn get_function_by_name(&self, name: &str, arguments: Vec<&str>) -> Option<FunctionMeta> {
+    pub fn get_function_by_name(
+        &self,
+        name: &str,
+        arguments: Vec<&str>,
+    ) -> Option<FunctionMeta<'a>> {
         unsafe {
             let name_ptr = name.as_ptr() as *const c_char;
             let arguments_str = arguments.join(",");
@@ -109,7 +117,7 @@ impl KeyspaceMeta {
     }
 
     /// Gets the aggregate metadata for the provided aggregate name.
-    pub fn aggregate_by_name(&self, name: &str, arguments: Vec<&str>) -> Option<AggregateMeta> {
+    pub fn aggregate_by_name(&self, name: &str, arguments: Vec<&str>) -> Option<AggregateMeta<'a>> {
         unsafe {
             let name_ptr = name.as_ptr() as *const c_char;
             let arguments_str = arguments.join(",");
@@ -130,17 +138,17 @@ impl KeyspaceMeta {
     }
 
     /// Iterator over the tables in this keyspaces
-    pub fn table_iter(&mut self) -> TableIterator {
+    pub fn table_iter(&self) -> TableIterator<'a> {
         unsafe { TableIterator::build(cass_iterator_tables_from_keyspace_meta(self.0)) }
     }
 
     /// Iterator over the functions in this keyspaces
-    pub fn function_iter(&mut self) -> FunctionIterator {
+    pub fn function_iter(&self) -> FunctionIterator<'a> {
         unsafe { FunctionIterator::build(cass_iterator_functions_from_keyspace_meta(self.0)) }
     }
 
     /// Iterator over the UDTs in this keyspaces
-    pub fn user_type_iter(&mut self) -> UserTypeIterator {
+    pub fn user_type_iter(&self) -> UserTypeIterator<'a> {
         unsafe { UserTypeIterator::build(cass_iterator_user_types_from_keyspace_meta(self.0)) }
     }
 
@@ -156,14 +164,14 @@ impl KeyspaceMeta {
 
     /// Gets a metadata field for the provided name. Metadata fields allow direct
     /// access to the column data found in the underlying "keyspaces" metadata table.
-    pub fn field_by_name(&self, name: &str) -> Option<MetadataFieldValue> {
+    pub fn field_by_name(&self, name: &str) -> Option<MetadataFieldValue<'a>> {
         unsafe {
             let name_ptr = name.as_ptr() as *const c_char;
             let value = cass_keyspace_meta_field_by_name_n(self.0, name_ptr, name.len());
             if value.is_null() {
                 None
             } else {
-                Some(MetadataFieldValue(value))
+                Some(MetadataFieldValue(value, PhantomData))
             }
         }
     }

@@ -9,7 +9,7 @@ use crate::cassandra_sys::cass_schema_meta_snapshot_version;
 use crate::cassandra_sys::CassSchemaMeta as _CassSchemaMeta;
 use std::os::raw::c_char;
 
-/// A snapshot of the schema's metadata
+/// A snapshot of the schema's metadata. Owned.
 #[derive(Debug)]
 pub struct SchemaMeta(*const _CassSchemaMeta);
 
@@ -56,7 +56,7 @@ impl SchemaMeta {
     }
 
     /// Returns an iterator over the keyspaces in this schema
-    pub fn keyspace_iter(&mut self) -> KeyspaceIterator {
+    pub fn keyspace_iter(&self) -> KeyspaceIterator {
         unsafe { KeyspaceIterator::build(cass_iterator_keyspaces_from_schema_meta(self.0)) }
     }
 }

@@ -16,44 +16,48 @@ use crate::cassandra_sys::cass_table_meta_name;
 use crate::cassandra_sys::cass_table_meta_partition_key;
 use crate::cassandra_sys::cass_table_meta_partition_key_count;
 use crate::cassandra_sys::CassTableMeta as _CassTableMeta;
-use std::mem;
+
+use std::marker::PhantomData;
+
 use std::os::raw::c_char;
 use std::slice;
 
 use std::str;
 
 /// Table metadata
+//
+// Borrowed from wherever the value is borrowed from.
 #[derive(Debug)]
-pub struct TableMeta(*const _CassTableMeta);
+pub struct TableMeta<'a>(*const _CassTableMeta, PhantomData<&'a _CassTableMeta>);
 
-impl ProtectedInner<*const _CassTableMeta> for TableMeta {
+impl ProtectedInner<*const _CassTableMeta> for TableMeta<'_> {
     fn inner(&self) -> *const _CassTableMeta {
         self.0
     }
 }
 
-impl Protected<*const _CassTableMeta> for TableMeta {
+impl Protected<*const _CassTableMeta> for TableMeta<'_> {
     fn build(inner: *const _CassTableMeta) -> Self {
         if inner.is_null() {
             panic!("Unexpected null pointer")
         };
-        TableMeta(inner)
+        TableMeta(inner, PhantomData)
     }
 }
 
-impl TableMeta {
+impl<'a> TableMeta<'a> {
     /// returns an iterator over the fields of this table
-    pub fn field_iter(&mut self) -> FieldIterator {
+    pub fn field_iter(&self) -> FieldIterator<'a> {
         unsafe { FieldIterator::build(cass_iterator_fields_from_table_meta(self.0)) }
     }
 
     /// An iterator over the columns in this table
-    pub fn columns_iter(&self) -> ColumnIterator {
+    pub fn columns_iter(&self) -> ColumnIterator<'a> {
         unsafe { ColumnIterator::build(cass_iterator_columns_from_table_meta(self.0)) }
     }
 
     /// Gets the column metadata for the provided column name.
-    pub fn column_by_name(&self, name: &str) -> ColumnMeta {
+    pub fn column_by_name(&self, name: &str) -> ColumnMeta<'a> {
         // TODO: can return NULL
         unsafe {
             ColumnMeta::build(cass_table_meta_column_by_name(
@@ -81,7 +85,7 @@ impl TableMeta {
     }
 
     /// Gets the column metadata for the provided index.
-    pub fn column(&self, index: usize) -> ColumnMeta {
+    pub fn column(&self, index: usize) -> ColumnMeta<'a> {
         // TODO: can return NULL
         unsafe { ColumnMeta::build(cass_table_meta_column(self.0, index)) }
     }
@@ -92,7 +96,7 @@ impl TableMeta {
     }
 
     /// Gets the partition key column metadata for the provided index.
-    pub fn partition_key(&self, index: usize) -> Option<ColumnMeta> {
+    pub fn partition_key(&self, index: usize) -> Option<ColumnMeta<'a>> {
         unsafe {
             let key = cass_table_meta_partition_key(self.0, index);
             if key.is_null() {
@@ -109,7 +113,7 @@ impl TableMeta {
     }
 
     /// Gets the clustering key column metadata for the provided index.
-    pub fn cluster_key(&self, index: usize) -> Option<ColumnMeta> {
+    pub fn cluster_key(&self, index: usize) -> Option<ColumnMeta<'a>> {
         unsafe {
             let key = cass_table_meta_clustering_key(self.0, index);
             if key.is_null() {
@@ -122,7 +126,7 @@ impl TableMeta {
 
     /// Gets a metadata field for the provided name. Metadata fields allow direct
     /// access to the column data found in the underlying "tables" metadata table.
-    pub fn field_by_name(&self, name: &str) -> Option<Value> {
+    pub fn field_by_name(&self, name: &str) -> Option<Value<'a>> {
         // fixme replace CassValule with a custom type
         unsafe {
             let value = cass_table_meta_field_by_name(self.0, name.as_ptr() as *const c_char);

@@ -2,7 +2,6 @@
 #![allow(dead_code)]
 #![allow(missing_copy_implementations)]
 
-use crate::cassandra::cluster::Cluster;
 use crate::cassandra::custom_payload::CustomPayloadResponse;
 use crate::cassandra::error::*;
 use crate::cassandra::future::CassFuture;
@@ -14,9 +13,6 @@ use crate::cassandra::statement::Statement;
 use crate::cassandra::util::{Protected, ProtectedInner};
 use crate::{cassandra::batch::Batch, BatchType};
 
-use crate::cassandra_sys::cass_session_close;
-use crate::cassandra_sys::cass_session_connect;
-use crate::cassandra_sys::cass_session_connect_keyspace_n;
 use crate::cassandra_sys::cass_session_execute;
 use crate::cassandra_sys::cass_session_execute_batch;
 use crate::cassandra_sys::cass_session_free;
@@ -26,7 +22,6 @@ use crate::cassandra_sys::cass_session_new;
 use crate::cassandra_sys::cass_session_prepare_n;
 use crate::cassandra_sys::CassSession as _Session;
 
-use std::ffi::NulError;
 use std::mem;
 use std::os::raw::c_char;
 use std::sync::Arc;

@@ -46,9 +46,10 @@ fn to_bitset(flags: &[SslVerifyFlag]) -> i32 {
 #[derive(Debug)]
 pub struct Ssl(*mut _Ssl);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
 unsafe impl Send for Ssl {}
+unsafe impl Sync for Ssl {}
 
 impl ProtectedInner<*mut _Ssl> for Ssl {
     fn inner(&self) -> *mut _Ssl {

@@ -1,5 +1,5 @@
 use crate::cassandra_sys::CASS_ERROR_LIB_INVALID_DATA;
-use bigdecimal::num_bigint::BigInt;
+
 use bigdecimal::BigDecimal;
 
 use crate::cassandra::collection::List;
@@ -95,9 +95,10 @@ impl StatementInner {
 #[derive(Debug)]
 pub struct Statement(StatementInner, Session);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
 unsafe impl Send for StatementInner {}
+unsafe impl Sync for StatementInner {}
 
 impl ProtectedInner<*mut _Statement> for StatementInner {
     #[inline(always)]

@@ -16,7 +16,7 @@ use crate::cassandra_sys::cass_tuple_new_from_data_type;
 use crate::cassandra_sys::cass_tuple_set_bool;
 use crate::cassandra_sys::cass_tuple_set_bytes;
 use crate::cassandra_sys::cass_tuple_set_collection;
-use crate::cassandra_sys::cass_tuple_set_decimal;
+
 use crate::cassandra_sys::cass_tuple_set_double;
 use crate::cassandra_sys::cass_tuple_set_float;
 use crate::cassandra_sys::cass_tuple_set_inet;
@@ -39,9 +39,10 @@ use std::os::raw::c_char;
 #[derive(Debug)]
 pub struct Tuple(*mut _Tuple);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
 unsafe impl Send for Tuple {}
+unsafe impl Sync for Tuple {}
 
 impl ProtectedInner<*mut _Tuple> for Tuple {
     fn inner(&self) -> *mut _Tuple {

@@ -16,8 +16,7 @@ use crate::cassandra_sys::cass_user_type_set_bytes;
 use crate::cassandra_sys::cass_user_type_set_bytes_by_name_n;
 use crate::cassandra_sys::cass_user_type_set_collection;
 use crate::cassandra_sys::cass_user_type_set_collection_by_name_n;
-use crate::cassandra_sys::cass_user_type_set_decimal;
-use crate::cassandra_sys::cass_user_type_set_decimal_by_name_n;
+
 use crate::cassandra_sys::cass_user_type_set_double;
 use crate::cassandra_sys::cass_user_type_set_double_by_name_n;
 use crate::cassandra_sys::cass_user_type_set_float;
@@ -53,9 +52,10 @@ use std::os::raw::c_char;
 #[derive(Debug)]
 pub struct UserType(*mut _UserType);
 
-// The underlying C type has no thread-local state, but does not support access
-// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
+// The underlying C type has no thread-local state, and forbids only concurrent
+// mutation/free: https://datastax.github.io/cpp-driver/topics/#thread-safety
 unsafe impl Send for UserType {}
+unsafe impl Sync for UserType {}
 
 impl ProtectedInner<*mut _UserType> for UserType {
     fn inner(&self) -> *mut _UserType {
@@ -78,12 +78,6 @@ impl Drop for UserType {
     }
 }
 
-// impl Drop for UserType {
-//    fn drop(&mut self) {unsafe{
-//        cass_user_type_free(self.0)
-//    }}
-// }
-
 impl UserType {
     /// Gets the data type of a user defined type.
     pub fn data_type(&self) -> ConstDataType {

@@ -21,7 +21,7 @@ use std::ffi::CStr;
 use std::fmt;
 use std::fmt::Formatter;
 use std::fmt::{Debug, Display};
-use std::mem;
+
 use std::os::raw::c_char;
 use std::str;
 

@@ -8,7 +8,7 @@ use crate::cassandra::util::{Protected, ProtectedInner};
 use crate::cassandra::uuid::Uuid;
 
 use crate::cassandra_sys::cass_bool_t;
-use crate::cassandra_sys::cass_collection_append_decimal;
+
 use crate::cassandra_sys::cass_iterator_fields_from_user_type;
 use crate::cassandra_sys::cass_iterator_from_collection;
 use crate::cassandra_sys::cass_iterator_from_map;
@@ -29,53 +29,24 @@ use crate::cassandra_sys::cass_value_get_uint32;
 use crate::cassandra_sys::cass_value_get_uuid;
 use crate::cassandra_sys::cass_value_is_collection;
 use crate::cassandra_sys::cass_value_is_null;
-use crate::cassandra_sys::cass_value_item_count;
-use crate::cassandra_sys::cass_value_primary_sub_type;
-use crate::cassandra_sys::cass_value_secondary_sub_type;
+
 use crate::cassandra_sys::cass_value_type;
 use crate::cassandra_sys::CassInet;
 use crate::cassandra_sys::CassUuid;
 use crate::cassandra_sys::CassValue as _CassValue;
 use crate::cassandra_sys::CassValueType_;
 use crate::cassandra_sys::CASS_ERROR_LIB_INVALID_VALUE_TYPE;
 use crate::cassandra_sys::CASS_ERROR_LIB_NULL_VALUE;
-use crate::cassandra_sys::CASS_VALUE_TYPE_ASCII;
-use crate::cassandra_sys::CASS_VALUE_TYPE_BIGINT;
-use crate::cassandra_sys::CASS_VALUE_TYPE_BLOB;
-use crate::cassandra_sys::CASS_VALUE_TYPE_BOOLEAN;
-use crate::cassandra_sys::CASS_VALUE_TYPE_COUNTER;
-use crate::cassandra_sys::CASS_VALUE_TYPE_CUSTOM;
-use crate::cassandra_sys::CASS_VALUE_TYPE_DATE;
-use crate::cassandra_sys::CASS_VALUE_TYPE_DECIMAL;
-use crate::cassandra_sys::CASS_VALUE_TYPE_DOUBLE;
-use crate::cassandra_sys::CASS_VALUE_TYPE_DURATION;
-use crate::cassandra_sys::CASS_VALUE_TYPE_FLOAT;
-use crate::cassandra_sys::CASS_VALUE_TYPE_INET;
-use crate::cassandra_sys::CASS_VALUE_TYPE_INT;
-use crate::cassandra_sys::CASS_VALUE_TYPE_LAST_ENTRY;
-use crate::cassandra_sys::CASS_VALUE_TYPE_LIST;
-use crate::cassandra_sys::CASS_VALUE_TYPE_MAP;
-use crate::cassandra_sys::CASS_VALUE_TYPE_SET;
-use crate::cassandra_sys::CASS_VALUE_TYPE_SMALL_INT;
-use crate::cassandra_sys::CASS_VALUE_TYPE_TEXT;
-use crate::cassandra_sys::CASS_VALUE_TYPE_TIME;
-use crate::cassandra_sys::CASS_VALUE_TYPE_TIMESTAMP;
-use crate::cassandra_sys::CASS_VALUE_TYPE_TIMEUUID;
-use crate::cassandra_sys::CASS_VALUE_TYPE_TINY_INT;
-use crate::cassandra_sys::CASS_VALUE_TYPE_TUPLE;
-use crate::cassandra_sys::CASS_VALUE_TYPE_UDT;
-use crate::cassandra_sys::CASS_VALUE_TYPE_UNKNOWN;
-use crate::cassandra_sys::CASS_VALUE_TYPE_UUID;
-use crate::cassandra_sys::CASS_VALUE_TYPE_VARCHAR;
-use crate::cassandra_sys::CASS_VALUE_TYPE_VARINT;
+
+use crate::LendingIterator;
 
 use bigdecimal::num_bigint::BigInt;
 use bigdecimal::BigDecimal;
-use std::ffi::CString;
+
 use std::fmt;
 use std::fmt::{Debug, Display, Formatter};
-use std::mem;
-use std::ptr;
+use std::marker::PhantomData;
+
 use std::slice;
 use std::str;
 
@@ -146,24 +117,27 @@ enhance_nullary_enum!(ValueType, CassValueType_, {
 }, omit { CASS_VALUE_TYPE_LAST_ENTRY });
 
 /// A single primitive value or a collection of values.
-pub struct Value(*const _CassValue);
+//
+// Borrowed immutably.
+pub struct Value<'a>(*const _CassValue, PhantomData<&'a _CassValue>);
 
 // The underlying C type is read-only so thread-safe.
-unsafe impl Send for Value {}
-unsafe impl Sync for Value {}
+// https://datastax.github.io/cpp-driver/topics/#thread-safety
+unsafe impl Send for Value<'_> {}
+unsafe impl Sync for Value<'_> {}
 
-impl ProtectedInner<*const _CassValue> for Value {
+impl ProtectedInner<*const _CassValue> for Value<'_> {
     fn inner(&self) -> *const _CassValue {
         self.0
     }
 }
 
-impl Protected<*const _CassValue> for Value {
+impl Protected<*const _CassValue> for Value<'_> {
     fn build(inner: *const _CassValue) -> Self {
         if inner.is_null() {
             panic!("Unexpected null pointer")
         };
-        Value(inner)
+        Value(inner, PhantomData)
     }
 }
 
@@ -175,8 +149,8 @@ where
     write!(f, "[")?;
     match set {
         Err(_) => write!(f, "<error>")?,
-        Ok(iter) => {
-            for item in iter {
+        Ok(mut iter) => {
+            while let Some(item) = iter.next() {
                 writer(f, item)?
             }
         }
@@ -193,8 +167,8 @@ where
     write!(f, "{{")?;
     match set {
         Err(_) => write!(f, "<error>")?,
-        Ok(iter) => {
-            for item in iter {
+        Ok(mut iter) => {
+            while let Some(item) = iter.next() {
                 writer(f, item.0, item.1)?
             }
         }
@@ -213,7 +187,7 @@ where
     }
 }
 
-impl Debug for Value {
+impl Debug for Value<'_> {
     fn fmt(&self, f: &mut Formatter) -> fmt::Result {
         if self.is_null() {
             Ok(())
@@ -259,7 +233,7 @@ impl Debug for Value {
     }
 }
 
-impl Display for Value {
+impl Display for Value<'_> {
     fn fmt(&self, f: &mut Formatter) -> fmt::Result {
         if self.is_null() {
             Ok(())
@@ -311,9 +285,9 @@ impl Display for Value {
     }
 }
 
-impl Value {
+impl<'a> Value<'a> {
     /// Get the raw bytes of this Cassandra value.
-    pub fn get_bytes(&self) -> Result<&[u8]> {
+    pub fn get_bytes(&self) -> Result<&'a [u8]> {
         let mut output = std::ptr::null();
         let mut output_size = 0;
         unsafe {
@@ -329,7 +303,7 @@ impl Value {
     }
 
     /// Get the data type of this Cassandra value
-    pub fn data_type(&self) -> ConstDataType {
+    pub fn data_type(&self) -> ConstDataType<'a> {
         unsafe { ConstDataType::build(cass_value_data_type(self.0)) }
     }
 
@@ -358,8 +332,8 @@ impl Value {
     //        unsafe { ValueType::build(cass_value_secondary_sub_type(self.0)).unwrap() }
     //    }
 
-    /// Gets this value as a set iterator.
-    pub fn get_set(&self) -> Result<SetIterator> {
+    /// Gets this value as a set / list / tuple iterator (the same method works for any of these).
+    pub fn get_set(&self) -> Result<SetIterator<'a>> {
         unsafe {
             match self.get_type() {
                 ValueType::SET | ValueType::LIST | ValueType::TUPLE => {
@@ -377,7 +351,7 @@ impl Value {
     }
 
     /// Gets this value as a map iterator.
-    pub fn get_map(&self) -> Result<MapIterator> {
+    pub fn get_map(&self) -> Result<MapIterator<'a>> {
         unsafe {
             match self.get_type() {
                 ValueType::MAP => {
@@ -395,7 +369,7 @@ impl Value {
     }
 
     /// Gets an iterator over the fields of the user type in this column or errors if you ask for the wrong type
-    pub fn get_user_type(&self) -> Result<UserTypeIterator> {
+    pub fn get_user_type(&self) -> Result<UserTypeIterator<'a>> {
         unsafe {
             match self.get_type() {
                 ValueType::UDT => {
@@ -413,7 +387,7 @@ impl Value {
     }
 
     /// Get this value as a string slice
-    pub fn get_str(&self) -> Result<&str> {
+    pub fn get_str(&self) -> Result<&'a str> {
         let mut message_ptr = std::ptr::null();
         let mut message_length = 0;
         unsafe {

@@ -1,4 +1,3 @@
-use crate::cassandra::util::Protected;
 use crate::cassandra_sys::CassWriteType_;
 
 /// A Cassandra write type level.

@@ -2,7 +2,6 @@
 #![deny(missing_docs)]
 #![allow(unknown_lints)]
 #![allow(clippy::doc_markdown)]
-#![allow(unused_imports)] // TODO: remove
 #![allow(dead_code)] // TODO: remove
 // `error_chain!` can recurse deeply
 #![recursion_limit = "1024"]
@@ -24,6 +23,7 @@ pub use crate::cassandra::data_type::DataType;
 // pub use cassandra::write_type::*;
 pub use crate::cassandra::field::Field;
 pub use crate::cassandra::inet::Inet;
+pub use crate::cassandra::iterator::LendingIterator;
 // pub use cassandra::util::*;
 // pub use cassandra::metrics::*;
 pub use crate::cassandra::iterator::{

@@ -121,10 +121,10 @@ fn basic_from_result(result: CassResult) -> Result<Option<Basic>> {
         None => Ok(None),
         Some(row) => {
             // todo: refactor?
-            let fields_iter: UserTypeIterator = row.get(12)?;
+            let mut fields_iter: UserTypeIterator = row.get(12)?;
             let mut dt: u32 = 0;
             let mut tm: i64 = 0;
-            for field in fields_iter {
+            while let Some(field) = fields_iter.next() {
                 match field.0.as_ref() {
                     "dt" => dt = field.1.get_u32()?,
                     "tm" => tm = field.1.get_i64()?,
@@ -183,7 +183,8 @@ async fn test_simple() -> Result<()> {
 
     println!("{}", result);
     let mut names = vec![];
-    for row in result.iter() {
+    let mut iter = result.iter();
+    while let Some(row) = iter.next() {
         let col: String = row.get_by_name("keyspace_name").unwrap();
         println!("ks name = {}", col);
         names.push(col);
@@ -338,7 +339,8 @@ async fn test_decimal_round_trip() -> Result<()> {
             ))
             .await?;
 
-        for row in result.into_iter() {
+        let mut iter = result.iter();
+        while let Some(row) = iter.next() {
             let txt: String = row.get_by_name("txt").unwrap();
             let dec: BigDecimal = row.get_by_name("dec").unwrap();
 
@@ -368,7 +370,8 @@ async fn test_decimal_round_trip() -> Result<()> {
             ))
             .await?;
 
-        for row in result.into_iter() {
+        let mut iter = result.iter();
+        while let Some(row) = iter.next() {
             let txt: String = row.get_by_name("txt").unwrap();
             let dec: BigDecimal = row.get_by_name("dec").unwrap();
 
@@ -631,7 +634,8 @@ async fn test_error_reporting() -> Result<()> {
     let result = session
         .execute("SELECT i32 FROM examples.basic WHERE key = 'utf8';")
         .await?;
-    let row = result.iter().next().unwrap();
+    let mut iter = result.iter();
+    let row = iter.next().unwrap();
     let err = row
         .get_column(0)?
         .get_string()

@@ -27,13 +27,14 @@ async fn insert_into_batch_with_prepared(session: &Session, pairs: &Vec<Pair>) -
 
 async fn retrieve_batch(session: &Session) -> Result<Vec<Pair>> {
     let result = session.execute("SELECT * from examples.pairs").await?;
-    let v = result
-        .iter()
-        .map(|r| Pair {
+    let mut v = vec![];
+    let mut iter = result.iter();
+    while let Some(r) = iter.next() {
+        v.push(Pair {
             key: r.get(0).expect("Key"),
             value: r.get(1).expect("Value"),
         })
-        .collect();
+    }
     Ok(v)
 }
 

@@ -19,7 +19,8 @@ async fn test_bind_by_name() -> Result<()> {
     for _ in 0..1000 {
         let statement = prepared_statement.bind();
         let result = statement.execute().await?;
-        for row in result.iter() {
+        let mut iter = result.iter();
+        while let Some(row) = iter.next() {
             let name: String = row.get_by_name("column_name")?;
             let ftype: String = row.get_by_name("type")?;
             // Actual values are not important; we're checking it doesn't panic or fail to return info.

@@ -31,10 +31,11 @@ async fn select_from_collections(session: &Session, key: &str) -> Result<Vec<Str
     let result = statement.execute().await?;
     println!("{:?}", result);
     let mut res = vec![];
-    for row in result.iter() {
+    let mut iter = result.iter();
+    while let Some(row) = iter.next() {
         let column = row.get_column(0);
-        let items_iterator: SetIterator = column?.get_set()?;
-        for item in items_iterator {
+        let mut items_iterator: SetIterator = column?.get_set()?;
+        while let Some(item) = items_iterator.next() {
             println!("item: {:?}", item);
             res.push(item.get_string().expect("Should exist").to_string());
         }

@@ -36,10 +36,11 @@ async fn select_from_maps(session: &Session, key: &str) -> Result<Vec<Pair>> {
     let result = statement.execute().await?;
     // println!("{:?}", result);
     let mut res = vec![];
-    for row in result.iter() {
+    let mut iter = result.iter();
+    while let Some(row) = iter.next() {
         let column = row.get_column(0).unwrap(); //FIXME
-        let items_iterator: MapIterator = column.get_map().unwrap();
-        for item in items_iterator {
+        let mut items_iterator: MapIterator = column.get_map().unwrap();
+        while let Some(item) = items_iterator.next() {
             println!("item: {:?}", item);
             res.push(Pair {
                 key: item.0.get_string().expect("key").to_string(),

@@ -46,7 +46,8 @@ async fn select_from_paging(session: &Session) -> Result<Vec<(String, String)>>
 
         let result = statement.execute().await?;
         println!("{:?}", result);
-        for row in result.iter() {
+        let mut iter = result.iter();
+        while let Some(row) = iter.next() {
             match row.get_column(0)?.get_string() {
                 Ok(key) => {
                     let key_str = key.to_string();
@@ -57,6 +58,7 @@ async fn select_from_paging(session: &Session) -> Result<Vec<(String, String)>>
                 Err(err) => panic!("{}", err),
             }
         }
+        drop(iter);
         has_more_pages = result.has_more_pages();
         if has_more_pages {
             prev_result = Some(result);

@@ -9,7 +9,8 @@ async fn execute_statement() -> Result<()> {
         .execute()
         .await?;
 
-    for row in result.iter() {
+    let mut iter = result.iter();
+    while let Some(row) = iter.next() {
         let col: String = row.get_by_name("keyspace_name")?;
         print!("ks = {}", col);
     }
@@ -27,7 +28,8 @@ async fn execute_prepared_statement() -> Result<()> {
     statement.bind_string(0, "key")?;
 
     let result = statement.execute().await?;
-    for row in result.iter() {
+    let mut iter = result.iter();
+    while let Some(row) = iter.next() {
         let col: String = row.get_by_name("value")?;
         print!("value = {}", col);
     }

@@ -27,14 +27,14 @@ async fn select_from_log(session: &Session, key: &str) -> Result<Vec<(Uuid, Stri
     let mut statement = session.statement(SELECT_QUERY);
     statement.bind(0, key)?;
     let results = statement.execute().await?;
-    Ok(results
-        .iter()
-        .map(|r| {
-            let t: Uuid = r.get_column(1).expect("time").get_uuid().expect("time");
-            let e: String = r.get(2).expect("entry");
-            (t, e)
-        })
-        .collect())
+    let mut vec = vec![];
+    let mut iter = results.iter();
+    while let Some(r) = iter.next() {
+        let t: Uuid = r.get_column(1).expect("time").get_uuid().expect("time");
+        let e: String = r.get(2).expect("entry");
+        vec.push((t, e));
+    }
+    Ok(vec)
 }
 
 #[tokio::test]