Merge 'Remove unsafe pointers (RawSlice) from RefValue' from Levy A.

- Add a lifetime parameter to `RefValue`, improving safety semantics by
preventing invalid pointers to `ImmutableRecord`.
- Also renames `RefValue` to `ValueRef`, to align with rusqlite and
other crates.
- Improve ergonomics by removing `RawSlice` in favor of native slices
Making it easier and safer to implement
https://github.com/tursodatabase/turso/issues/2304.
`TextSubtype` is stored as part of the enum variant of `ValueRef::Text`,
but this will be changed in a more general reworking of subtyping
described in https://github.com/tursodatabase/turso/issues/3573

Reviewed-by: Preston Thorpe <preston@turso.tech>

Closes #3587

core/incremental/aggregate_operator.rs

@@ -7,7 +7,7 @@ use crate::incremental::operator::{
     generate_storage_id, ComputationTracker, DbspStateCursors, EvalState, IncrementalOperator,
 };
 use crate::incremental::persistence::{ReadRecord, WriteRow};
-use crate::types::{IOResult, ImmutableRecord, RefValue, SeekKey, SeekOp, SeekResult};
+use crate::types::{IOResult, ImmutableRecord, SeekKey, SeekOp, SeekResult, ValueRef};
 use crate::{return_and_restore_if_io, return_if_io, LimboError, Result, Value};
 use std::collections::{BTreeMap, HashMap, HashSet};
 use std::fmt::{self, Display};
@@ -1546,7 +1546,7 @@ impl ScanState {
         };
 
         // Check if we're still in the same group
-        if let RefValue::Integer(rec_sid) = rec_storage_id {
+        if let ValueRef::Integer(rec_sid) = rec_storage_id {
             if *rec_sid != storage_id {
                 return Ok(IOResult::Done(None));
             }
@@ -1555,8 +1555,8 @@ impl ScanState {
         }
 
         // Compare zset_hash as blob
-        if let RefValue::Blob(rec_zset_blob) = rec_zset_hash {
-            if let Some(rec_hash) = Hash128::from_blob(rec_zset_blob.to_slice()) {
+        if let ValueRef::Blob(rec_zset_blob) = rec_zset_hash {
+            if let Some(rec_hash) = Hash128::from_blob(rec_zset_blob) {
                 if rec_hash != zset_hash {
                     return Ok(IOResult::Done(None));
                 }

core/incremental/cursor.rs

@@ -117,15 +117,12 @@ impl MaterializedViewCursor {
             Some(rowid) => rowid,
         };
 
-        let btree_record = return_if_io!(self.btree_cursor.record());
-        let btree_ref_values = btree_record
-            .ok_or_else(|| {
-                crate::LimboError::InternalError(
-                    "Invalid data in materialized view: found a rowid, but not the row!"
-                        .to_string(),
-                )
-            })?
-            .get_values();
+        let btree_record = return_if_io!(self.btree_cursor.record()).ok_or_else(|| {
+            crate::LimboError::InternalError(
+                "Invalid data in materialized view: found a rowid, but not the row!".to_string(),
+            )
+        })?;
+        let btree_ref_values = btree_record.get_values();
 
         // Convert RefValues to Values (copying for now - can optimize later)
         let mut btree_values: Vec<Value> =

core/json/mod.rs

@@ -11,9 +11,9 @@ pub use crate::json::ops::{
     jsonb_replace,
 };
 use crate::json::path::{json_path, JsonPath, PathElement};
-use crate::types::{RawSlice, Text, TextRef, TextSubtype, Value, ValueType};
+use crate::types::{Text, TextSubtype, Value, ValueType};
 use crate::vdbe::Register;
-use crate::{bail_constraint_error, bail_parse_error, LimboError, RefValue};
+use crate::{bail_constraint_error, bail_parse_error, LimboError, ValueRef};
 pub use cache::JsonCacheCell;
 use jsonb::{ElementType, Jsonb, JsonbHeader, PathOperationMode, SearchOperation, SetOperation};
 use std::borrow::Cow;
@@ -105,14 +105,12 @@ pub fn json_from_raw_bytes_agg(data: &[u8], raw: bool) -> crate::Result<Value> {
 
 pub fn convert_dbtype_to_jsonb(val: &Value, strict: Conv) -> crate::Result<Jsonb> {
     convert_ref_dbtype_to_jsonb(
-        &match val {
-            Value::Null => RefValue::Null,
-            Value::Integer(x) => RefValue::Integer(*x),
-            Value::Float(x) => RefValue::Float(*x),
-            Value::Text(text) => {
-                RefValue::Text(TextRef::create_from(text.as_str().as_bytes(), text.subtype))
-            }
-            Value::Blob(items) => RefValue::Blob(RawSlice::create_from(items)),
+        match val {
+            Value::Null => ValueRef::Null,
+            Value::Integer(x) => ValueRef::Integer(*x),
+            Value::Float(x) => ValueRef::Float(*x),
+            Value::Text(text) => ValueRef::Text(text.as_str().as_bytes(), text.subtype),
+            Value::Blob(items) => ValueRef::Blob(items.as_slice()),
         },
         strict,
     )
@@ -124,14 +122,14 @@ fn parse_as_json_text(slice: &[u8]) -> crate::Result<Jsonb> {
     Jsonb::from_str_with_mode(str, Conv::Strict).map_err(Into::into)
 }
 
-pub fn convert_ref_dbtype_to_jsonb(val: &RefValue, strict: Conv) -> crate::Result<Jsonb> {
+pub fn convert_ref_dbtype_to_jsonb(val: ValueRef<'_>, strict: Conv) -> crate::Result<Jsonb> {
     match val {
-        RefValue::Text(text) => {
-            let res = if text.subtype == TextSubtype::Json || matches!(strict, Conv::Strict) {
-                Jsonb::from_str_with_mode(text.as_str(), strict)
+        ValueRef::Text(text, subtype) => {
+            let res = if subtype == TextSubtype::Json || matches!(strict, Conv::Strict) {
+                Jsonb::from_str_with_mode(&String::from_utf8_lossy(text), strict)
             } else {
                 // Handle as a string literal otherwise
-                let mut str = text.as_str().replace('"', "\\\"");
+                let mut str = String::from_utf8_lossy(text).replace('"', "\\\"");
                 // Quote the string to make it a JSON string
                 str.insert(0, '"');
                 str.push('"');
@@ -139,8 +137,8 @@ pub fn convert_ref_dbtype_to_jsonb(val: &RefValue, strict: Conv) -> crate::Resul
             };
             res.map_err(|_| LimboError::ParseError("malformed JSON".to_string()))
         }
-        RefValue::Blob(blob) => {
-            let bytes = blob.to_slice();
+        ValueRef::Blob(blob) => {
+            let bytes = blob;
             // Valid JSON can start with these whitespace characters
             let index = bytes
                 .iter()
@@ -177,15 +175,15 @@ pub fn convert_ref_dbtype_to_jsonb(val: &RefValue, strict: Conv) -> crate::Resul
             json.element_type()?;
             Ok(json)
         }
-        RefValue::Null => Ok(Jsonb::from_raw_data(
+        ValueRef::Null => Ok(Jsonb::from_raw_data(
             JsonbHeader::make_null().into_bytes().as_bytes(),
         )),
-        RefValue::Float(float) => {
+        ValueRef::Float(float) => {
             let mut buff = ryu::Buffer::new();
-            Jsonb::from_str(buff.format(*float))
+            Jsonb::from_str(buff.format(float))
                 .map_err(|_| LimboError::ParseError("malformed JSON".to_string()))
         }
-        RefValue::Integer(int) => Jsonb::from_str(&int.to_string())
+        ValueRef::Integer(int) => Jsonb::from_str(&int.to_string())
             .map_err(|_| LimboError::ParseError("malformed JSON".to_string())),
     }
 }

core/lib.rs

@@ -97,8 +97,8 @@ use turso_macros::match_ignore_ascii_case;
 use turso_parser::ast::fmt::ToTokens;
 use turso_parser::{ast, ast::Cmd, parser::Parser};
 use types::IOResult;
-pub use types::RefValue;
 pub use types::Value;
+pub use types::ValueRef;
 use util::parse_schema_rows;
 pub use util::IOExt;
 pub use vdbe::{builder::QueryMode, explain::EXPLAIN_COLUMNS, explain::EXPLAIN_QUERY_PLAN_COLUMNS};

core/mvcc/database/checkpoint_state_machine.rs

@@ -9,8 +9,8 @@ use crate::storage::pager::CreateBTreeFlags;
 use crate::storage::wal::{CheckpointMode, TursoRwLock};
 use crate::types::{IOCompletions, IOResult, ImmutableRecord, RecordCursor};
 use crate::{
-    CheckpointResult, Completion, Connection, IOExt, Pager, RefValue, Result, TransactionState,
-    Value,
+    CheckpointResult, Completion, Connection, IOExt, Pager, Result, TransactionState, Value,
+    ValueRef,
 };
 use parking_lot::RwLock;
 use std::collections::{HashMap, HashSet};
@@ -191,7 +191,7 @@ impl<Clock: LogicalClock> CheckpointStateMachine<Clock> {
                         let row_data = ImmutableRecord::from_bin_record(row_data.clone());
                         let mut record_cursor = RecordCursor::new();
                         record_cursor.parse_full_header(&row_data).unwrap();
-                        let RefValue::Integer(root_page) =
+                        let ValueRef::Integer(root_page) =
                             record_cursor.get_value(&row_data, 3).unwrap()
                         else {
                             panic!(

core/mvcc/database/mod.rs

@@ -18,11 +18,11 @@ use crate::Completion;
 use crate::File;
 use crate::IOExt;
 use crate::LimboError;
-use crate::RefValue;
 use crate::Result;
 use crate::Statement;
 use crate::StepResult;
 use crate::Value;
+use crate::ValueRef;
 use crate::{Connection, Pager};
 use crossbeam_skiplist::{SkipMap, SkipSet};
 use parking_lot::RwLock;
@@ -1978,7 +1978,7 @@ impl<Clock: LogicalClock> MvStore<Clock> {
                         let record = ImmutableRecord::from_bin_record(row_data);
                         let mut record_cursor = RecordCursor::new();
                         let record_values = record_cursor.get_values(&record).unwrap();
-                        let RefValue::Integer(root_page) = record_values[3] else {
+                        let ValueRef::Integer(root_page) = record_values[3] else {
                             panic!(
                                 "Expected integer value for root page, got {:?}",
                                 record_values[3]

core/mvcc/database/tests.rs

@@ -714,7 +714,7 @@ use crate::types::Text;
 use crate::Value;
 use crate::{Database, StepResult};
 use crate::{MemoryIO, Statement};
-use crate::{RefValue, DATABASE_MANAGER};
+use crate::{ValueRef, DATABASE_MANAGER};
 
 // Simple atomic clock implementation for testing
 
@@ -978,8 +978,8 @@ fn test_cursor_modification_during_scan() {
     record.start_serialization(&row.data);
     let value = record.get_value(0).unwrap();
     match value {
-        RefValue::Text(text) => {
-            assert_eq!(text.as_str(), "new_row");
+        ValueRef::Text(text, _) => {
+            assert_eq!(text, b"new_row");
         }
         _ => panic!("Expected Text value"),
     }
@@ -1210,8 +1210,8 @@ fn test_restart() {
             .unwrap();
         let record = get_record_value(&row);
         match record.get_value(0).unwrap() {
-            RefValue::Text(text) => {
-                assert_eq!(text.as_str(), "bar");
+            ValueRef::Text(text, _) => {
+                assert_eq!(text, b"bar");
             }
             _ => panic!("Expected Text value"),
         }

core/mvcc/persistent_storage/logical_log.rs

@@ -501,7 +501,7 @@ mod tests {
             LocalClock, MvStore,
         },
         types::{ImmutableRecord, Text},
-        OpenFlags, RefValue, Value,
+        OpenFlags, Value, ValueRef,
     };
 
     use super::LogRecordType;
@@ -565,10 +565,10 @@ mod tests {
         let record = ImmutableRecord::from_bin_record(row.data.clone());
         let values = record.get_values();
         let foo = values.first().unwrap();
-        let RefValue::Text(foo) = foo else {
+        let ValueRef::Text(foo, _) = foo else {
             unreachable!()
         };
-        assert_eq!(foo.as_str(), "foo");
+        assert_eq!(foo, b"foo");
     }
 
     #[test]
@@ -637,10 +637,10 @@ mod tests {
             let record = ImmutableRecord::from_bin_record(row.data.clone());
             let values = record.get_values();
             let foo = values.first().unwrap();
-            let RefValue::Text(foo) = foo else {
+            let ValueRef::Text(foo, _) = foo else {
                 unreachable!()
             };
-            assert_eq!(foo.as_str(), value.as_str());
+            assert_eq!(*foo, value.as_bytes());
         }
     }
 
@@ -758,11 +758,14 @@ mod tests {
             let record = ImmutableRecord::from_bin_record(row.data.clone());
             let values = record.get_values();
             let foo = values.first().unwrap();
-            let RefValue::Text(foo) = foo else {
+            let ValueRef::Text(foo, _) = foo else {
                 unreachable!()
             };
 
-            assert_eq!(foo.as_str(), format!("row_{}", present_rowid.row_id as u64));
+            assert_eq!(
+                String::from_utf8_lossy(foo),
+                format!("row_{}", present_rowid.row_id as u64)
+            );
         }
 
         // Check rowids that were deleted

core/schema.rs

@@ -80,7 +80,7 @@ use crate::util::{
 };
 use crate::{
     bail_parse_error, contains_ignore_ascii_case, eq_ignore_ascii_case, match_ignore_ascii_case,
-    Connection, LimboError, MvCursor, MvStore, Pager, RefValue, SymbolTable, VirtualTable,
+    Connection, LimboError, MvCursor, MvStore, Pager, SymbolTable, ValueRef, VirtualTable,
 };
 use crate::{util::normalize_ident, Result};
 use core::fmt;
@@ -428,36 +428,36 @@ impl Schema {
             let mut record_cursor = cursor.record_cursor.borrow_mut();
             // sqlite schema table has 5 columns: type, name, tbl_name, rootpage, sql
             let ty_value = record_cursor.get_value(&row, 0)?;
-            let RefValue::Text(ty) = ty_value else {
+            let ValueRef::Text(ty, _) = ty_value else {
                 return Err(LimboError::ConversionError("Expected text value".into()));
             };
-            let ty = ty.as_str();
-            let RefValue::Text(name) = record_cursor.get_value(&row, 1)? else {
+            let ty = String::from_utf8_lossy(ty);
+            let ValueRef::Text(name, _) = record_cursor.get_value(&row, 1)? else {
                 return Err(LimboError::ConversionError("Expected text value".into()));
             };
-            let name = name.as_str();
+            let name = String::from_utf8_lossy(name);
             let table_name_value = record_cursor.get_value(&row, 2)?;
-            let RefValue::Text(table_name) = table_name_value else {
+            let ValueRef::Text(table_name, _) = table_name_value else {
                 return Err(LimboError::ConversionError("Expected text value".into()));
             };
-            let table_name = table_name.as_str();
+            let table_name = String::from_utf8_lossy(table_name);
             let root_page_value = record_cursor.get_value(&row, 3)?;
-            let RefValue::Integer(root_page) = root_page_value else {
+            let ValueRef::Integer(root_page) = root_page_value else {
                 return Err(LimboError::ConversionError("Expected integer value".into()));
             };
             let sql_value = record_cursor.get_value(&row, 4)?;
             let sql_textref = match sql_value {
-                RefValue::Text(sql) => Some(sql),
+                ValueRef::Text(sql, _) => Some(sql),
                 _ => None,
             };
-            let sql = sql_textref.as_ref().map(|s| s.as_str());
+            let sql = sql_textref.map(|s| String::from_utf8_lossy(s));
 
             self.handle_schema_row(
-                ty,
-                name,
-                table_name,
+                &ty,
+                &name,
+                &table_name,
                 root_page,
-                sql,
+                sql.as_deref(),
                 syms,
                 &mut from_sql_indexes,
                 &mut automatic_indices,

core/storage/btree.rs

@@ -28,7 +28,7 @@ use crate::{
 
 use crate::{
     return_corrupt, return_if_io,
-    types::{compare_immutable, IOResult, ImmutableRecord, RefValue, SeekKey, SeekOp, Value},
+    types::{compare_immutable, IOResult, ImmutableRecord, SeekKey, SeekOp, Value, ValueRef},
     LimboError, Result,
 };
 
@@ -705,7 +705,7 @@ impl BTreeCursor {
             .unwrap()
             .last_value(record_cursor)
         {
-            Some(Ok(RefValue::Integer(rowid))) => rowid,
+            Some(Ok(ValueRef::Integer(rowid))) => rowid,
             _ => unreachable!(
                 "index where has_rowid() is true should have an integer rowid as the last value"
             ),
@@ -2164,7 +2164,7 @@ impl BTreeCursor {
 
     fn compare_with_current_record(
         &self,
-        key_values: &[RefValue],
+        key_values: &[ValueRef],
         seek_op: SeekOp,
         record_comparer: &RecordCompare,
         index_info: &IndexInfo,
@@ -8690,7 +8690,11 @@ mod tests {
                 run_until_done(|| cursor.next(), pager.deref()).unwrap();
                 let record = run_until_done(|| cursor.record(), &pager).unwrap();
                 let record = record.as_ref().unwrap();
-                let cur = record.get_values().clone();
+                let cur = record
+                    .get_values()
+                    .iter()
+                    .map(ValueRef::to_owned)
+                    .collect::<Vec<_>>();
                 if let Some(prev) = prev {
                     if prev >= cur {
                         println!("Seed: {seed}");
@@ -8930,14 +8934,10 @@ mod tests {
             let record = record.as_ref().unwrap();
             let cur = record.get_values().clone();
             let cur = cur.first().unwrap();
-            let RefValue::Blob(ref cur) = cur else {
+            let ValueRef::Blob(ref cur) = cur else {
                 panic!("expected blob, got {cur:?}");
             };
-            assert_eq!(
-                cur.to_slice(),
-                key,
-                "key {key:?} is not found, seed: {seed}"
-            );
+            assert_eq!(cur, key, "key {key:?} is not found, seed: {seed}");
         }
         pager.end_read_tx();
     }
@@ -9469,11 +9469,11 @@ mod tests {
             let exists = run_until_done(|| cursor.next(), &pager)?;
             assert!(exists, "Record {i} not found");
 
-            let record = run_until_done(|| cursor.record(), &pager)?;
-            let value = record.unwrap().get_value(0)?;
+            let record = run_until_done(|| cursor.record(), &pager)?.unwrap();
+            let value = record.get_value(0)?;
             assert_eq!(
                 value,
-                RefValue::Integer(i),
+                ValueRef::Integer(i),
                 "Unexpected value for record {i}",
             );
         }

core/storage/sqlite3_ondisk.rs

@@ -61,7 +61,7 @@ use crate::storage::buffer_pool::BufferPool;
 use crate::storage::database::{DatabaseFile, DatabaseStorage, EncryptionOrChecksum};
 use crate::storage::pager::Pager;
 use crate::storage::wal::READMARK_NOT_USED;
-use crate::types::{RawSlice, RefValue, SerialType, SerialTypeKind, TextRef, TextSubtype};
+use crate::types::{SerialType, SerialTypeKind, TextSubtype, ValueRef};
 use crate::{
     bail_corrupt_error, turso_assert, CompletionError, File, IOContext, Result, WalFileShared,
 };
@@ -1320,22 +1320,22 @@ impl<T: Default + Copy, const N: usize> Iterator for SmallVecIter<'_, T, N> {
 /// Reads a value that might reference the buffer it is reading from. Be sure to store RefValue with the buffer
 /// always.
 #[inline(always)]
-pub fn read_value(buf: &[u8], serial_type: SerialType) -> Result<(RefValue, usize)> {
+pub fn read_value<'a>(buf: &'a [u8], serial_type: SerialType) -> Result<(ValueRef<'a>, usize)> {
     match serial_type.kind() {
-        SerialTypeKind::Null => Ok((RefValue::Null, 0)),
+        SerialTypeKind::Null => Ok((ValueRef::Null, 0)),
         SerialTypeKind::I8 => {
             if buf.is_empty() {
                 crate::bail_corrupt_error!("Invalid UInt8 value");
             }
             let val = buf[0] as i8;
-            Ok((RefValue::Integer(val as i64), 1))
+            Ok((ValueRef::Integer(val as i64), 1))
         }
         SerialTypeKind::I16 => {
             if buf.len() < 2 {
                 crate::bail_corrupt_error!("Invalid BEInt16 value");
             }
             Ok((
-                RefValue::Integer(i16::from_be_bytes([buf[0], buf[1]]) as i64),
+                ValueRef::Integer(i16::from_be_bytes([buf[0], buf[1]]) as i64),
                 2,
             ))
         }
@@ -1345,7 +1345,7 @@ pub fn read_value(buf: &[u8], serial_type: SerialType) -> Result<(RefValue, usiz
             }
             let sign_extension = if buf[0] <= 127 { 0 } else { 255 };
             Ok((
-                RefValue::Integer(
+                ValueRef::Integer(
                     i32::from_be_bytes([sign_extension, buf[0], buf[1], buf[2]]) as i64
                 ),
                 3,
@@ -1356,7 +1356,7 @@ pub fn read_value(buf: &[u8], serial_type: SerialType) -> Result<(RefValue, usiz
                 crate::bail_corrupt_error!("Invalid BEInt32 value");
             }
             Ok((
-                RefValue::Integer(i32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]) as i64),
+                ValueRef::Integer(i32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]) as i64),
                 4,
             ))
         }
@@ -1366,7 +1366,7 @@ pub fn read_value(buf: &[u8], serial_type: SerialType) -> Result<(RefValue, usiz
             }
             let sign_extension = if buf[0] <= 127 { 0 } else { 255 };
             Ok((
-                RefValue::Integer(i64::from_be_bytes([
+                ValueRef::Integer(i64::from_be_bytes([
                     sign_extension,
                     sign_extension,
                     buf[0],
@@ -1384,7 +1384,7 @@ pub fn read_value(buf: &[u8], serial_type: SerialType) -> Result<(RefValue, usiz
                 crate::bail_corrupt_error!("Invalid BEInt64 value");
             }
             Ok((
-                RefValue::Integer(i64::from_be_bytes([
+                ValueRef::Integer(i64::from_be_bytes([
                     buf[0], buf[1], buf[2], buf[3], buf[4], buf[5], buf[6], buf[7],
                 ])),
                 8,
@@ -1395,26 +1395,20 @@ pub fn read_value(buf: &[u8], serial_type: SerialType) -> Result<(RefValue, usiz
                 crate::bail_corrupt_error!("Invalid BEFloat64 value");
             }
             Ok((
-                RefValue::Float(f64::from_be_bytes([
+                ValueRef::Float(f64::from_be_bytes([
                     buf[0], buf[1], buf[2], buf[3], buf[4], buf[5], buf[6], buf[7],
                 ])),
                 8,
             ))
         }
-        SerialTypeKind::ConstInt0 => Ok((RefValue::Integer(0), 0)),
-        SerialTypeKind::ConstInt1 => Ok((RefValue::Integer(1), 0)),
+        SerialTypeKind::ConstInt0 => Ok((ValueRef::Integer(0), 0)),
+        SerialTypeKind::ConstInt1 => Ok((ValueRef::Integer(1), 0)),
         SerialTypeKind::Blob => {
             let content_size = serial_type.size();
             if buf.len() < content_size {
                 crate::bail_corrupt_error!("Invalid Blob value");
             }
-            if content_size == 0 {
-                Ok((RefValue::Blob(RawSlice::new(std::ptr::null(), 0)), 0))
-            } else {
-                let ptr = &buf[0] as *const u8;
-                let slice = RawSlice::new(ptr, content_size);
-                Ok((RefValue::Blob(slice), content_size))
-            }
+            Ok((ValueRef::Blob(&buf[..content_size]), content_size))
         }
         SerialTypeKind::Text => {
             let content_size = serial_type.size();
@@ -1427,10 +1421,7 @@ pub fn read_value(buf: &[u8], serial_type: SerialType) -> Result<(RefValue, usiz
             }
 
             Ok((
-                RefValue::Text(TextRef::create_from(
-                    &buf[..content_size],
-                    TextSubtype::Text,
-                )),
+                ValueRef::Text(&buf[..content_size], TextSubtype::Text),
                 content_size,
             ))
         }

core/vdbe/execute.rs

@@ -65,7 +65,7 @@ use crate::{
     vector::{vector32, vector64, vector_distance_cos, vector_distance_l2, vector_extract},
 };
 
-use crate::{info, turso_assert, OpenFlags, RefValue, Row, TransactionState};
+use crate::{info, turso_assert, OpenFlags, Row, TransactionState, ValueRef};
 
 use super::{
     insn::{Cookie, RegisterOrLiteral},
@@ -2705,7 +2705,7 @@ pub fn op_row_id(
                     let record_cursor = record_cursor_ref.deref_mut();
                     let rowid = record.last_value(record_cursor).unwrap();
                     match rowid {
-                        Ok(RefValue::Integer(rowid)) => rowid,
+                        Ok(ValueRef::Integer(rowid)) => rowid,
                         _ => unreachable!(),
                     }
                 };
@@ -4275,7 +4275,14 @@ pub fn op_sorter_compare(
         &record.get_values()[..*num_regs]
     };
 
-    let cursor = state.get_cursor(*cursor_id);
+    // Inlined `state.get_cursor` to prevent borrowing conflit with `state.registers`
+    let cursor = state
+        .cursors
+        .get_mut(*cursor_id)
+        .unwrap_or_else(|| panic!("cursor id {cursor_id} out of bounds"))
+        .as_mut()
+        .unwrap_or_else(|| panic!("cursor id {cursor_id} is None"));
+
     let cursor = cursor.as_sorter_mut();
     let Some(current_sorter_record) = cursor.record() else {
         return Err(LimboError::InternalError(
@@ -4287,7 +4294,7 @@ pub fn op_sorter_compare(
     // If the current sorter record has a NULL in any of the significant fields, the comparison is not equal.
     let is_equal = current_sorter_values
         .iter()
-        .all(|v| !matches!(v, RefValue::Null))
+        .all(|v| !matches!(v, ValueRef::Null))
         && compare_immutable(
             previous_sorter_values,
             current_sorter_values,
@@ -4953,7 +4960,7 @@ pub fn op_function(
                 }
                 #[cfg(feature = "json")]
                 {
-                    use crate::types::{TextRef, TextSubtype};
+                    use crate::types::TextSubtype;
 
                     let table = state.registers[*start_reg].get_value();
                     let Value::Text(table) = table else {
@@ -4978,10 +4985,7 @@ pub fn op_function(
                     for column in table.columns() {
                         let name = column.name.as_ref().unwrap();
                         let name_json = json::convert_ref_dbtype_to_jsonb(
-                            &RefValue::Text(TextRef::create_from(
-                                name.as_str().as_bytes(),
-                                TextSubtype::Text,
-                            )),
+                            ValueRef::Text(name.as_bytes(), TextSubtype::Text),
                             json::Conv::ToString,
                         )?;
                         json.append_jsonb_to_end(name_json.data());
@@ -5049,13 +5053,13 @@ pub fn op_function(
                         json.append_jsonb_to_end(column_name.data());
 
                         let val = record_cursor.get_value(&record, i)?;
-                        if let RefValue::Blob(..) = val {
+                        if let ValueRef::Blob(..) = val {
                             return Err(LimboError::InvalidArgument(
                                 "bin_record_json_object: formatting of BLOB values stored in binary record is not supported".to_string()
                             ));
                         }
                         let val_json =
-                            json::convert_ref_dbtype_to_jsonb(&val, json::Conv::NotStrict)?;
+                            json::convert_ref_dbtype_to_jsonb(val, json::Conv::NotStrict)?;
                         json.append_jsonb_to_end(val_json.data());
                     }
                     json.finalize_unsafe(json::jsonb::ElementType::OBJECT)?;
@@ -6249,9 +6253,9 @@ pub fn op_idx_insert(
                 // UNIQUE indexes disallow duplicates like (a=1,b=2,rowid=1) and (a=1,b=2,rowid=2).
                 let existing_key = if cursor.has_rowid() {
                     let count = cursor.record_cursor.borrow_mut().count(record);
-                    record.get_values()[..count.saturating_sub(1)].to_vec()
+                    &record.get_values()[..count.saturating_sub(1)]
                 } else {
-                    record.get_values().to_vec()
+                    &record.get_values()[..]
                 };
                 let inserted_key_vals = &record_to_insert.get_values();
                 if existing_key.len() != inserted_key_vals.len() {
@@ -6259,7 +6263,7 @@ pub fn op_idx_insert(
                 }
 
                 let conflict = compare_immutable(
-                    existing_key.as_slice(),
+                    existing_key,
                     inserted_key_vals,
                     &cursor.index_info.as_ref().unwrap().key_info,
                 ) == std::cmp::Ordering::Equal;
@@ -6550,7 +6554,7 @@ pub fn op_no_conflict(
                         record
                             .get_values()
                             .iter()
-                            .any(|val| matches!(val, RefValue::Null))
+                            .any(|val| matches!(val, ValueRef::Null))
                     }
                     RecordSource::Unpacked {
                         start_reg,

core/vdbe/mod.rs

@@ -32,7 +32,7 @@ use crate::{
     state_machine::StateMachine,
     storage::{pager::PagerCommitResult, sqlite3_ondisk::SmallVec},
     translate::{collate::CollationSeq, plan::TableReferences},
-    types::{IOCompletions, IOResult, RawSlice, TextRef},
+    types::{IOCompletions, IOResult},
     vdbe::{
         execute::{
             OpCheckpointState, OpColumnState, OpDeleteState, OpDeleteSubState, OpDestroyState,
@@ -41,7 +41,7 @@ use crate::{
         },
         metrics::StatementMetrics,
     },
-    RefValue,
+    ValueRef,
 };
 
 use crate::{
@@ -1001,22 +1001,10 @@ fn make_record(registers: &[Register], start_reg: &usize, count: &usize) -> Immu
     ImmutableRecord::from_registers(regs, regs.len())
 }
 
-pub fn registers_to_ref_values(registers: &[Register]) -> Vec<RefValue> {
+pub fn registers_to_ref_values<'a>(registers: &'a [Register]) -> Vec<ValueRef<'a>> {
     registers
         .iter()
-        .map(|reg| {
-            let value = reg.get_value();
-            match value {
-                Value::Null => RefValue::Null,
-                Value::Integer(i) => RefValue::Integer(*i),
-                Value::Float(f) => RefValue::Float(*f),
-                Value::Text(t) => RefValue::Text(TextRef {
-                    value: RawSlice::new(t.value.as_ptr(), t.value.len()),
-                    subtype: t.subtype,
-                }),
-                Value::Blob(b) => RefValue::Blob(RawSlice::new(b.as_ptr(), b.len())),
-            }
-        })
+        .map(|reg| reg.get_value().as_ref())
         .collect()
 }
 

core/vdbe/sorter.rs

@@ -1,6 +1,6 @@
 use turso_parser::ast::SortOrder;
 
-use std::cell::RefCell;
+use std::cell::{RefCell, UnsafeCell};
 use std::cmp::{Eq, Ord, Ordering, PartialEq, PartialOrd, Reverse};
 use std::collections::BinaryHeap;
 use std::rc::Rc;
@@ -15,7 +15,7 @@ use crate::{
     storage::sqlite3_ondisk::{read_varint, varint_len, write_varint},
     translate::collate::CollationSeq,
     turso_assert,
-    types::{IOResult, ImmutableRecord, KeyInfo, RecordCursor, RefValue},
+    types::{IOResult, ImmutableRecord, KeyInfo, RecordCursor, ValueRef},
     Result,
 };
 use crate::{io_yield_many, io_yield_one, return_if_io, CompletionError};
@@ -614,7 +614,8 @@ impl SortedChunk {
 struct SortableImmutableRecord {
     record: ImmutableRecord,
     cursor: RecordCursor,
-    key_values: RefCell<Vec<RefValue>>,
+    // SAFETY: borrows from 'self
+    key_values: UnsafeCell<Vec<ValueRef<'static>>>,
     key_len: usize,
     index_key_info: Rc<Vec<KeyInfo>>,
     /// The key deserialization error, if any.
@@ -636,29 +637,34 @@ impl SortableImmutableRecord {
         Ok(Self {
             record,
             cursor,
-            key_values: RefCell::new(Vec::with_capacity(key_len)),
+            key_values: UnsafeCell::new(Vec::with_capacity(key_len)),
             index_key_info,
             deserialization_error: RefCell::new(None),
             key_len,
         })
     }
 
-    /// Attempts to deserialize the key value at the given index.
-    /// If the key value has already been deserialized, this does nothing.
-    /// The deserialized key value is stored in the `key_values` field.
-    /// In case of an error, the error is stored in the `deserialization_error` field.
-    fn try_deserialize_key(&self, idx: usize) {
-        let mut key_values = self.key_values.borrow_mut();
-        if idx < key_values.len() {
-            // The key value with this index has already been deserialized.
-            return;
-        }
-        match self.cursor.deserialize_column(&self.record, idx) {
-            Ok(value) => key_values.push(value),
-            Err(error) => {
-                self.deserialization_error.replace(Some(error));
-            }
+    fn key_value<'a>(&'a self, i: usize) -> Option<ValueRef<'a>> {
+        // SAFETY: there are no other active references
+        let key_values = unsafe { &mut *self.key_values.get() };
+
+        if i >= key_values.len() {
+            assert_eq!(key_values.len(), i, "access must be sequential");
+
+            let value = match self.cursor.deserialize_column(&self.record, i) {
+                Ok(value) => value,
+                Err(err) => {
+                    self.deserialization_error.replace(Some(err));
+                    return None;
+                }
+            };
+
+            // SAFETY: no 'static lifetime is exposed, all references are bound to 'self
+            let value: ValueRef<'static> = unsafe { std::mem::transmute(value) };
+            key_values.push(value);
         }
+
+        Some(key_values[i])
     }
 }
 
@@ -674,34 +680,25 @@ impl Ord for SortableImmutableRecord {
             self.cursor.serial_types.len(),
             other.cursor.serial_types.len()
         );
-        let this_key_values_len = self.key_values.borrow().len();
-        let other_key_values_len = other.key_values.borrow().len();
 
         for i in 0..self.key_len {
-            // Lazily deserialize the key values if they haven't been deserialized already.
-            if i >= this_key_values_len {
-                self.try_deserialize_key(i);
-                if self.deserialization_error.borrow().is_some() {
-                    return Ordering::Equal;
-                }
-            }
-            if i >= other_key_values_len {
-                other.try_deserialize_key(i);
-                if other.deserialization_error.borrow().is_some() {
-                    return Ordering::Equal;
-                }
-            }
+            let Some(this_key_value) = self.key_value(i) else {
+                return Ordering::Equal;
+            };
+
+            let Some(other_key_value) = other.key_value(i) else {
+                return Ordering::Equal;
+            };
 
-            let this_key_value = &self.key_values.borrow()[i];
-            let other_key_value = &other.key_values.borrow()[i];
             let column_order = self.index_key_info[i].sort_order;
             let collation = self.index_key_info[i].collation;
 
             let cmp = match (this_key_value, other_key_value) {
-                (RefValue::Text(left), RefValue::Text(right)) => {
-                    collation.compare_strings(left.as_str(), right.as_str())
-                }
-                _ => this_key_value.partial_cmp(other_key_value).unwrap(),
+                (ValueRef::Text(left, _), ValueRef::Text(right, _)) => collation.compare_strings(
+                    &String::from_utf8_lossy(left),
+                    &String::from_utf8_lossy(right),
+                ),
+                _ => this_key_value.partial_cmp(&other_key_value).unwrap(),
             };
             if !cmp.is_eq() {
                 return match column_order {
@@ -742,7 +739,7 @@ enum SortedChunkIOState {
 mod tests {
     use super::*;
     use crate::translate::collate::CollationSeq;
-    use crate::types::{ImmutableRecord, RefValue, Value, ValueType};
+    use crate::types::{ImmutableRecord, Value, ValueRef, ValueType};
     use crate::util::IOExt;
     use crate::PlatformIO;
     use rand_chacha::{
@@ -806,7 +803,7 @@ mod tests {
             for i in 0..num_records {
                 assert!(sorter.has_more());
                 let record = sorter.record().unwrap();
-                assert_eq!(record.get_values()[0], RefValue::Integer(i));
+                assert_eq!(record.get_values()[0], ValueRef::Integer(i));
                 // Check that the record remained unchanged after sorting.
                 assert_eq!(record, &initial_records[(num_records - i - 1) as usize]);