mvcc: table cursor must always use defined bound

because all tables are stored in the same skipmap, so omitting bound
for search will find rows from wrong table

Author: jussisaurio

core/mvcc/cursor.rs

@@ -252,7 +252,10 @@ impl<Clock: LogicalClock + 'static> MvccLazyCursor<Clock> {
                 in_btree: false,
                 btree_consumed: true,
             },
-            (None, None) => CursorPosition::End,
+            (None, None) => match direction {
+                IterationDirection::Forwards => CursorPosition::BeforeFirst,
+                IterationDirection::Backwards => CursorPosition::End,
+            },
         }
     }
 
@@ -319,14 +322,22 @@ impl<Clock: LogicalClock + 'static> CursorTrait for MvccLazyCursor<Clock> {
         let current_state = self.state.borrow().clone();
         if current_state.is_none() {
             let before_first = matches!(self.get_current_pos(), CursorPosition::BeforeFirst);
-            let min_key_exclusive = match self.current_pos.borrow().clone() {
+            let min_key = match self.current_pos.borrow().clone() {
                 CursorPosition::Loaded {
                     row_id,
                     in_btree: _,
                     btree_consumed: _,
-                } => Some(row_id.row_id),
+                } => Some(match &self.mv_cursor_type {
+                    MvccCursorType::Table => {
+                        (true, RowKey::Int(row_id.row_id.to_int_or_panic() + 1))
+                    }
+                    MvccCursorType::Index(_) => (false, row_id.row_id),
+                }),
                 // TODO: do we need to forward twice?
-                CursorPosition::BeforeFirst => None,
+                CursorPosition::BeforeFirst => match &self.mv_cursor_type {
+                    MvccCursorType::Table => Some((false, RowKey::Int(i64::MIN))),
+                    MvccCursorType::Index(_) => None,
+                },
                 CursorPosition::End => {
                     // let's keep same state, we reached the end so no point in moving forward.
                     return Ok(IOResult::Done(false));
@@ -335,7 +346,7 @@ impl<Clock: LogicalClock + 'static> CursorTrait for MvccLazyCursor<Clock> {
 
             let new_position_in_mvcc = match self.db.get_next_row_id_for_table(
                 self.table_id,
-                min_key_exclusive.as_ref(),
+                min_key.as_ref().map(|(inclusive, key)| (*inclusive, key)),
                 self.tx_id,
             ) {
                 Some(id) => CursorPosition::Loaded {
@@ -455,21 +466,29 @@ impl<Clock: LogicalClock + 'static> CursorTrait for MvccLazyCursor<Clock> {
     fn prev(&mut self) -> Result<IOResult<bool>> {
         let current_state = self.state.borrow().clone();
         if current_state.is_none() {
-            let max_id_exclusive = match self.current_pos.borrow().clone() {
+            let max_key = match self.current_pos.borrow().clone() {
                 CursorPosition::Loaded {
                     row_id,
                     in_btree: _,
                     btree_consumed: _,
-                } => Some(row_id.row_id.clone()),
+                } => Some(match &self.mv_cursor_type {
+                    MvccCursorType::Table => {
+                        (true, RowKey::Int(row_id.row_id.to_int_or_panic() - 1))
+                    }
+                    MvccCursorType::Index(_) => (false, row_id.row_id),
+                }),
                 CursorPosition::BeforeFirst => {
                     return Ok(IOResult::Done(false));
                 }
-                CursorPosition::End => None,
+                CursorPosition::End => match &self.mv_cursor_type {
+                    MvccCursorType::Table => Some((true, RowKey::Int(i64::MAX))),
+                    MvccCursorType::Index(_) => None,
+                },
             };
 
             let new_position_in_mvcc = match self.db.get_prev_row_id_for_table(
                 self.table_id,
-                max_id_exclusive.as_ref(),
+                max_key.as_ref().map(|(inclusive, key)| (*inclusive, key)),
                 self.tx_id,
             ) {
                 Some(id) => CursorPosition::Loaded {
@@ -802,9 +821,18 @@ impl<Clock: LogicalClock + 'static> CursorTrait for MvccLazyCursor<Clock> {
             self.current_pos.replace(CursorPosition::BeforeFirst);
         }
 
-        let new_position_in_mvcc =
-            self.db
-                .get_next_row_id_for_table(self.table_id, None, self.tx_id);
+        let start_key_mvcc = match &self.mv_cursor_type {
+            MvccCursorType::Table => Some((true, RowKey::Int(i64::MIN))),
+            MvccCursorType::Index(_) => None,
+        };
+
+        let new_position_in_mvcc = self.db.get_next_row_id_for_table(
+            self.table_id,
+            start_key_mvcc
+                .as_ref()
+                .map(|(inclusive, key)| (*inclusive, key)),
+            self.tx_id,
+        );
 
         let maybe_rowid_in_btree = match &self.mv_cursor_type {
             MvccCursorType::Table => {
@@ -828,7 +856,7 @@ impl<Clock: LogicalClock + 'static> CursorTrait for MvccLazyCursor<Clock> {
             }
         };
         let new_position = self.get_new_position_from_mvcc_and_btree(
-            &new_position_in_mvcc.map(|r| r.row_id),
+            &new_position_in_mvcc.clone().map(|r| r.row_id),
             &maybe_rowid_in_btree,
             IterationDirection::Forwards,
         );

core/mvcc/database/mod.rs

@@ -1401,7 +1401,7 @@ impl<Clock: LogicalClock> MvStore<Clock> {
     pub fn get_next_row_id_for_table(
         &self,
         table_id: MVTableId,
-        start: Option<&RowKey>,
+        start: Option<(bool, &RowKey)>, // (inclusive, row_key)
         tx_id: TxID,
     ) -> Option<RowID> {
         let res = self.get_row_id_for_table_in_direction(
@@ -1423,7 +1423,7 @@ impl<Clock: LogicalClock> MvStore<Clock> {
     pub fn get_prev_row_id_for_table(
         &self,
         table_id: MVTableId,
-        start: Option<&RowKey>,
+        start: Option<(bool, &RowKey)>, // (inclusive, row_key)
         tx_id: TxID,
     ) -> Option<RowID> {
         let res = self.get_row_id_for_table_in_direction(
@@ -1445,15 +1445,15 @@ impl<Clock: LogicalClock> MvStore<Clock> {
     pub fn get_row_id_for_table_in_direction(
         &self,
         table_id: MVTableId,
-        start: Option<&RowKey>,
+        start: Option<(bool, &RowKey)>, // (inclusive, row_key)
         tx_id: TxID,
         direction: IterationDirection,
     ) -> Option<RowID> {
         tracing::trace!(
             "getting_row_id_for_table_in_direction(table_id={}, range_start={:?}, direction={:?})",
             table_id,
             start,
-            direction
+            direction,
         );
 
         let tx = self
@@ -1464,13 +1464,16 @@ impl<Clock: LogicalClock> MvStore<Clock> {
         if direction == IterationDirection::Forwards {
             let min_bound = start.map(|start| RowID {
                 table_id,
-                row_id: start.clone(),
+                row_id: start.1.clone(),
             });
+            let inclusive = start.map(|start| start.0).unwrap_or(false);
 
             match min_bound {
                 Some(min_bound) => {
                     let mut rows = self.rows.range(min_bound..);
-                    rows.next(); // exclusive wrt. min bound
+                    if !inclusive {
+                        rows.next();
+                    }
                     loop {
                         // We are moving forward, so if a row was deleted we just need to skip it. Therefore, we need
                         // to loop either until we find a row that is not deleted or until we reach the end of the table.
@@ -1519,13 +1522,16 @@ impl<Clock: LogicalClock> MvStore<Clock> {
         } else {
             let max_bound = start.map(|start| RowID {
                 table_id,
-                row_id: start.clone(),
+                row_id: start.1.clone(),
             });
+            let inclusive = start.map(|start| start.0).unwrap_or(false);
 
             match max_bound {
                 Some(max_bound) => {
-                    let mut rows = self.rows.range(max_bound..).rev();
-                    rows.next(); // exclusive wrt. max bound
+                    let mut rows = self.rows.range(..=max_bound).rev();
+                    if !inclusive {
+                        rows.next();
+                    }
                     loop {
                         // We are moving backwards, so if a row was deleted we just need to skip it. Therefore, we need
                         // to loop either until we find a row that is not deleted or until we reach the beginning of the table.