Fix UnexpectedNullError when deserializing (NULL,NotNull) into Option

diesel/src/mysql/connection/stmt/iterator.rs

@@ -59,6 +59,10 @@ impl<'a> Row<Mysql> for MysqlRow<'a> {
     fn next_is_null(&self, count: usize) -> bool {
         (0..count).all(|i| self.binds.field_data(self.col_idx + i).is_none())
     }
+
+    fn column_index(&self) -> usize {
+        self.col_idx
+    }
 }
 
 pub struct NamedStatementIterator<'a> {

diesel/src/pg/connection/row.rs

@@ -31,6 +31,10 @@ impl<'a> Row<Pg> for PgRow<'a> {
     fn next_is_null(&self, count: usize) -> bool {
         (0..count).all(|i| self.db_result.is_null(self.row_idx, self.col_idx + i))
     }
+
+    fn column_index(&self) -> usize {
+        self.col_idx
+    }
 }
 
 pub struct PgNamedRow<'a> {

diesel/src/row.rs

@@ -19,6 +19,12 @@ pub trait Row<DB: Backend> {
     /// would all return `None`.
     fn next_is_null(&self, count: usize) -> bool;
 
+    /// Returns the index of the current column (next to be returned by take, zero based)
+    ///
+    /// May be used to skip the right number of fields when recovering for an `UnexpectedNullError` when
+    /// deserializing an `Option`
+    fn column_index(&self) -> usize;
+
     /// Skips the next `count` columns. This method must be called if you are
     /// choosing not to call `take` as a result of `next_is_null` returning
     /// `true`.

diesel/src/sqlite/connection/functions.rs

@@ -30,7 +30,7 @@ where
     }
 
     conn.register_sql_function(fn_name, fields_needed, deterministic, move |conn, args| {
-        let mut row = FunctionRow { args };
+        let mut row = FunctionRow { args, col_idx: 0 };
         let args_row = Args::Row::build_from_row(&mut row).map_err(Error::DeserializationError)?;
         let args = Args::build(args_row);
 
@@ -55,11 +55,13 @@ where
 
 struct FunctionRow<'a> {
     args: &'a [*mut ffi::sqlite3_value],
+    col_idx: usize,
 }
 
 impl<'a> Row<Sqlite> for FunctionRow<'a> {
     fn take(&mut self) -> Option<&SqliteValue> {
         self.args.split_first().and_then(|(&first, rest)| {
+            self.col_idx += 1;
             self.args = rest;
             unsafe { SqliteValue::new(first) }
         })
@@ -70,4 +72,8 @@ impl<'a> Row<Sqlite> for FunctionRow<'a> {
             .iter()
             .all(|&p| unsafe { SqliteValue::new(p) }.is_none())
     }
+
+    fn column_index(&self) -> usize {
+        self.col_idx
+    }
 }

diesel/src/sqlite/connection/sqlite_value.rs

@@ -105,6 +105,10 @@ impl Row<Sqlite> for SqliteRow {
             tpe == ffi::SQLITE_NULL
         })
     }
+
+    fn column_index(&self) -> usize {
+        self.next_col_index as usize
+    }
 }
 
 pub struct SqliteNamedRow<'a> {

diesel/src/type_impls/option.rs

@@ -95,7 +95,27 @@ where
             row.advance(fields_needed);
             Ok(None)
         } else {
-            T::build_from_row(row).map(Some)
+            let column_index_before_build = row.column_index();
+            match T::build_from_row(row) {
+                Ok(v) => Ok(Some(v)),
+                Err(e) => {
+                    if e.is::<UnexpectedNullError>() {
+                        // It is possible that we are recovering but not all fields have been taken.
+                        // We need to skip those that are left that belong to this `Option`.
+
+                        // We are returning a deserialization error in case this goes unexpectedly
+                        // to make it easier on people implementing `FromSqlRow`
+                        row.advance(
+                            (column_index_before_build + fields_needed)
+                                .checked_sub(row.column_index())
+                                .ok_or("Inconsistent FromSqlRow impl: we went further than FIELDS_NEEDED")?,
+                        );
+                        Ok(None)
+                    } else {
+                        Err(e)
+                    }
+                }
+            }
         }
     }
 }

diesel_tests/tests/joins.rs

@@ -300,6 +300,35 @@ fn select_right_side_with_nullable_column_first() {
     assert_eq!(expected_data, actual_data);
 }
 
+#[test]
+fn select_left_join_right_side_with_non_null_inside() {
+    let connection = connection_with_sean_and_tess_in_users_table();
+
+    connection
+        .execute(
+            "INSERT INTO posts (user_id, title, body) VALUES
+        (1, 'Hello', 'Content')
+    ",
+        )
+        .unwrap();
+
+    let expected_data = vec![
+        (None, 2),
+        (Some((1, "Hello".to_string(), "Hello".to_string())), 1),
+    ];
+
+    let source = users::table
+        .left_outer_join(posts::table)
+        .select((
+            (users::id, posts::title, posts::title).nullable(),
+            users::id,
+        ))
+        .order_by((users::id.desc(), posts::id.asc()));
+    let actual_data: Vec<_> = source.load(&connection).unwrap();
+
+    assert_eq!(expected_data, actual_data);
+}
+
 #[test]
 fn select_then_join() {
     use crate::schema::users::dsl::*;