wip

bindings/go/rs_src/types.rs

@@ -143,23 +143,21 @@ impl LimboValue {
         Box::into_raw(Box::new(self)) as *const c_void
     }
 
-    pub fn from_value(value: &limbo_core::OwnedValue) -> Self {
+    pub fn from_value(value: &limbo_core::RefValue) -> Self {
         match value {
-            limbo_core::OwnedValue::Integer(i) => {
+            limbo_core::RefValue::Integer(i) => {
                 LimboValue::new(ValueType::Integer, ValueUnion::from_int(*i))
             }
-            limbo_core::OwnedValue::Float(r) => {
+            limbo_core::RefValue::Float(r) => {
                 LimboValue::new(ValueType::Real, ValueUnion::from_real(*r))
             }
-            limbo_core::OwnedValue::Text(s) => {
+            limbo_core::RefValue::Text(s) => {
                 LimboValue::new(ValueType::Text, ValueUnion::from_str(s.as_str()))
             }
-            limbo_core::OwnedValue::Blob(b) => {
-                LimboValue::new(ValueType::Blob, ValueUnion::from_bytes(b))
-            }
-            limbo_core::OwnedValue::Null => {
-                LimboValue::new(ValueType::Null, ValueUnion::from_null())
+            limbo_core::RefValue::Blob(b) => {
+                LimboValue::new(ValueType::Blob, ValueUnion::from_bytes(b.to_slice()))
             }
+            limbo_core::RefValue::Null => LimboValue::new(ValueType::Null, ValueUnion::from_null()),
         }
     }
 

bindings/java/rs_src/limbo_statement.rs

@@ -107,15 +107,15 @@ fn row_to_obj_array<'local>(
 
     for (i, value) in row.get_values().iter().enumerate() {
         let obj = match value {
-            limbo_core::OwnedValue::Null => JObject::null(),
-            limbo_core::OwnedValue::Integer(i) => {
+            limbo_core::RefValue::Null => JObject::null(),
+            limbo_core::RefValue::Integer(i) => {
                 env.new_object("java/lang/Long", "(J)V", &[JValue::Long(*i)])?
             }
-            limbo_core::OwnedValue::Float(f) => {
+            limbo_core::RefValue::Float(f) => {
                 env.new_object("java/lang/Double", "(D)V", &[JValue::Double(*f)])?
             }
-            limbo_core::OwnedValue::Text(s) => env.new_string(s.as_str())?.into(),
-            limbo_core::OwnedValue::Blob(b) => env.byte_array_from_slice(&b)?.into(),
+            limbo_core::RefValue::Text(s) => env.new_string(s.as_str())?.into(),
+            limbo_core::RefValue::Blob(b) => env.byte_array_from_slice(&b.to_slice())?.into(),
         };
         if let Err(e) = env.set_object_array_element(&obj_array, i as i32, obj) {
             eprintln!("Error on parsing row: {:?}", e);

bindings/javascript/src/lib.rs

@@ -92,14 +92,14 @@ impl Statement {
     }
 }
 
-fn to_js_value(env: &napi::Env, value: &limbo_core::OwnedValue) -> JsUnknown {
+fn to_js_value(env: &napi::Env, value: &limbo_core::RefValue) -> JsUnknown {
     match value {
-        limbo_core::OwnedValue::Null => env.get_null().unwrap().into_unknown(),
-        limbo_core::OwnedValue::Integer(i) => env.create_int64(*i).unwrap().into_unknown(),
-        limbo_core::OwnedValue::Float(f) => env.create_double(*f).unwrap().into_unknown(),
-        limbo_core::OwnedValue::Text(s) => env.create_string(s.as_str()).unwrap().into_unknown(),
-        limbo_core::OwnedValue::Blob(b) => {
-            env.create_buffer_copy(b.as_ref()).unwrap().into_unknown()
+        limbo_core::RefValue::Null => env.get_null().unwrap().into_unknown(),
+        limbo_core::RefValue::Integer(i) => env.create_int64(*i).unwrap().into_unknown(),
+        limbo_core::RefValue::Float(f) => env.create_double(*f).unwrap().into_unknown(),
+        limbo_core::RefValue::Text(s) => env.create_string(s.as_str()).unwrap().into_unknown(),
+        limbo_core::RefValue::Blob(b) => {
+            env.create_buffer_copy(b.to_slice()).unwrap().into_unknown()
         }
     }
 }

bindings/python/src/lib.rs

@@ -326,13 +326,11 @@ fn row_to_py(py: Python, row: &limbo_core::Row) -> Result<PyObject> {
     let mut py_values = Vec::new();
     for value in row.get_values() {
         match value {
-            limbo_core::OwnedValue::Null => py_values.push(py.None()),
-            limbo_core::OwnedValue::Integer(i) => py_values.push(i.into_pyobject(py)?.into()),
-            limbo_core::OwnedValue::Float(f) => py_values.push(f.into_pyobject(py)?.into()),
-            limbo_core::OwnedValue::Text(s) => py_values.push(s.as_str().into_pyobject(py)?.into()),
-            limbo_core::OwnedValue::Blob(b) => {
-                py_values.push(PyBytes::new(py, b.as_slice()).into())
-            }
+            limbo_core::RefValue::Null => py_values.push(py.None()),
+            limbo_core::RefValue::Integer(i) => py_values.push(i.into_pyobject(py)?.into()),
+            limbo_core::RefValue::Float(f) => py_values.push(f.into_pyobject(py)?.into()),
+            limbo_core::RefValue::Text(s) => py_values.push(s.as_str().into_pyobject(py)?.into()),
+            limbo_core::RefValue::Blob(b) => py_values.push(PyBytes::new(py, b.to_slice()).into()),
         }
     }
     Ok(PyTuple::new(py, &py_values)

bindings/rust/src/lib.rs

@@ -212,7 +212,7 @@ impl Rows {
             Ok(limbo_core::StepResult::Row) => {
                 let row = stmt.row().unwrap();
                 Ok(Some(Row {
-                    values: row.get_values().to_vec(),
+                    values: row.get_values().iter().map(|v| v.to_owned()).collect(),
                 }))
             }
             _ => Ok(None),

bindings/wasm/lib.rs

@@ -177,10 +177,10 @@ impl Statement {
     }
 }
 
-fn to_js_value(value: &limbo_core::OwnedValue) -> JsValue {
+fn to_js_value(value: &limbo_core::RefValue) -> JsValue {
     match value {
-        limbo_core::OwnedValue::Null => JsValue::null(),
-        limbo_core::OwnedValue::Integer(i) => {
+        limbo_core::RefValue::Null => JsValue::null(),
+        limbo_core::RefValue::Integer(i) => {
             let i = *i;
             if i >= i32::MIN as i64 && i <= i32::MAX as i64 {
                 JsValue::from(i as i32)
@@ -188,9 +188,9 @@ fn to_js_value(value: &limbo_core::OwnedValue) -> JsValue {
                 JsValue::from(i)
             }
         }
-        limbo_core::OwnedValue::Float(f) => JsValue::from(*f),
-        limbo_core::OwnedValue::Text(t) => JsValue::from_str(t.as_str()),
-        limbo_core::OwnedValue::Blob(b) => js_sys::Uint8Array::from(b.as_slice()).into(),
+        limbo_core::RefValue::Float(f) => JsValue::from(*f),
+        limbo_core::RefValue::Text(t) => JsValue::from_str(t.as_str()),
+        limbo_core::RefValue::Blob(b) => js_sys::Uint8Array::from(b.to_slice()).into(),
     }
 }
 

cli/app.rs

@@ -5,7 +5,7 @@ use crate::{
     opcodes_dictionary::OPCODE_DESCRIPTIONS,
 };
 use comfy_table::{Attribute, Cell, CellAlignment, Color, ContentArrangement, Row, Table};
-use limbo_core::{Database, LimboError, OwnedValue, Statement, StepResult};
+use limbo_core::{Database, LimboError, OwnedValue, RefValue, Statement, StepResult};
 
 use clap::{Parser, ValueEnum};
 use rustyline::{history::DefaultHistory, Editor};
@@ -769,19 +769,19 @@ impl<'a> Limbo<'a> {
                                 row.max_height(1);
                                 for (idx, value) in record.get_values().iter().enumerate() {
                                     let (content, alignment) = match value {
-                                        OwnedValue::Null => {
+                                        RefValue::Null => {
                                             (self.opts.null_value.clone(), CellAlignment::Left)
                                         }
-                                        OwnedValue::Integer(_) => {
+                                        RefValue::Integer(_) => {
                                             (format!("{}", value), CellAlignment::Right)
                                         }
-                                        OwnedValue::Float(_) => {
+                                        RefValue::Float(_) => {
                                             (format!("{}", value), CellAlignment::Right)
                                         }
-                                        OwnedValue::Text(_) => {
+                                        RefValue::Text(_) => {
                                             (format!("{}", value), CellAlignment::Left)
                                         }
-                                        OwnedValue::Blob(_) => {
+                                        RefValue::Blob(_) => {
                                             (format!("{}", value), CellAlignment::Left)
                                         }
                                     };
@@ -849,7 +849,7 @@ impl<'a> Limbo<'a> {
                     match rows.step()? {
                         StepResult::Row => {
                             let row = rows.row().unwrap();
-                            if let Some(OwnedValue::Text(schema)) = row.get_values().first() {
+                            if let Some(RefValue::Text(schema)) = row.get_values().first() {
                                 let _ = self.write_fmt(format_args!("{};", schema.as_str()));
                                 found = true;
                             }
@@ -907,7 +907,7 @@ impl<'a> Limbo<'a> {
                     match rows.step()? {
                         StepResult::Row => {
                             let row = rows.row().unwrap();
-                            if let Some(OwnedValue::Text(table)) = row.get_values().first() {
+                            if let Some(RefValue::Text(table)) = row.get_values().first() {
                                 tables.push_str(table.as_str());
                                 tables.push(' ');
                             }

core/lib.rs

@@ -63,6 +63,7 @@ use storage::{
 };
 use translate::select::prepare_select_plan;
 pub use types::OwnedValue;
+pub use types::RefValue;
 use util::{columns_from_create_table_body, parse_schema_rows};
 use vdbe::{builder::QueryMode, VTabOpaqueCursor};
 
@@ -596,7 +597,7 @@ impl Statement {
     }
 }
 
-pub type Row = types::Record;
+pub type Row = types::ImmutableRecord;
 
 pub type StepResult = vdbe::StepResult;
 

core/pseudo.rs

@@ -1,7 +1,7 @@
-use crate::types::Record;
+use crate::types::ImmutableRecord;
 
 pub struct PseudoCursor {
-    current: Option<Record>,
+    current: Option<ImmutableRecord>,
 }
 
 impl PseudoCursor {
@@ -9,11 +9,11 @@ impl PseudoCursor {
         Self { current: None }
     }
 
-    pub fn record(&self) -> Option<&Record> {
+    pub fn record(&self) -> Option<&ImmutableRecord> {
         self.current.as_ref()
     }
 
-    pub fn insert(&mut self, record: Record) {
+    pub fn insert(&mut self, record: ImmutableRecord) {
         self.current = Some(record);
     }
 }

core/storage/btree.rs

@@ -7,8 +7,8 @@ use crate::storage::sqlite3_ondisk::{
 use crate::MvCursor;
 
 use crate::types::{
-    compare_immutable_to_record, compare_record_to_immutable, CursorResult, ImmutableRecord,
-    OwnedValue, Record, RefValue, SeekKey, SeekOp,
+    compare_immutable, compare_immutable_to_record, CursorResult, ImmutableRecord, OwnedValue,
+    RefValue, SeekKey, SeekOp,
 };
 use crate::{return_corrupt, LimboError, Result};
 
@@ -610,8 +610,7 @@ impl BTreeCursor {
                     let SeekKey::IndexKey(index_key) = key else {
                         unreachable!("index seek key should be a record");
                     };
-                    let order =
-                        compare_immutable_to_record(&record.get_values(), &index_key.get_values());
+                    let order = compare_immutable(&record.get_values(), index_key.get_values());
                     let found = match op {
                         SeekOp::GT => order.is_gt(),
                         SeekOp::GE => order.is_ge(),
@@ -654,8 +653,7 @@ impl BTreeCursor {
                     let SeekKey::IndexKey(index_key) = key else {
                         unreachable!("index seek key should be a record");
                     };
-                    let order =
-                        compare_immutable_to_record(&record.get_values(), &index_key.get_values());
+                    let order = compare_immutable(&record.get_values(), index_key.get_values());
                     let found = match op {
                         SeekOp::GT => order.is_lt(),
                         SeekOp::GE => order.is_le(),
@@ -753,7 +751,7 @@ impl BTreeCursor {
                         } else {
                             crate::storage::sqlite3_ondisk::read_record(payload)?
                         };
-                        let order = compare_immutable_to_record(
+                        let order = compare_immutable(
                             &record.get_values().as_slice()[..record.len() - 1],
                             &index_key.get_values().as_slice()[..],
                         );
@@ -939,10 +937,7 @@ impl BTreeCursor {
                         } else {
                             crate::storage::sqlite3_ondisk::read_record(payload)?
                         };
-                        let order = compare_record_to_immutable(
-                            &index_key.get_values(),
-                            &record.get_values(),
-                        );
+                        let order = compare_immutable(index_key.get_values(), record.get_values());
                         let target_leaf_page_is_in_the_left_subtree = match cmp {
                             SeekOp::GT => order.is_lt(),
                             SeekOp::GE => order.is_le(),
@@ -984,7 +979,11 @@ impl BTreeCursor {
 
     /// Insert a record into the btree.
     /// If the insert operation overflows the page, it will be split and the btree will be balanced.
-    fn insert_into_page(&mut self, key: &OwnedValue, record: &Record) -> Result<CursorResult<()>> {
+    fn insert_into_page(
+        &mut self,
+        key: &OwnedValue,
+        record: &ImmutableRecord,
+    ) -> Result<CursorResult<()>> {
         if let CursorState::None = &self.state {
             self.state = CursorState::Write(WriteInfo::new());
         }
@@ -1943,7 +1942,7 @@ impl BTreeCursor {
     pub fn insert(
         &mut self,
         key: &OwnedValue,
-        record: &Record,
+        record: &ImmutableRecord,
         moved_before: bool, /* Indicate whether it's necessary to traverse to find the leaf page */
     ) -> Result<CursorResult<()>> {
         let int_key = match key {
@@ -1954,8 +1953,7 @@ impl BTreeCursor {
             Some(mv_cursor) => {
                 let row_id =
                     crate::mvcc::database::RowID::new(self.table_id() as u64, *int_key as u64);
-                let mut record_buf = Vec::new();
-                record.serialize(&mut record_buf);
+                let record_buf = record.payload.to_vec();
                 let row = crate::mvcc::database::Row::new(row_id, record_buf);
                 mv_cursor.borrow_mut().insert(row).unwrap();
             }
@@ -2594,7 +2592,7 @@ impl BTreeCursor {
         &mut self,
         page_ref: PageRef,
         cell_idx: usize,
-        record: &Record,
+        record: &ImmutableRecord,
     ) -> Result<CursorResult<()>> {
         // build the new payload
         let page_type = page_ref.get().contents.as_ref().unwrap().page_type();
@@ -3354,7 +3352,7 @@ fn fill_cell_payload(
     page_type: PageType,
     int_key: Option<u64>,
     cell_payload: &mut Vec<u8>,
-    record: &Record,
+    record: &ImmutableRecord,
     usable_space: u16,
     pager: Rc<Pager>,
 ) {
@@ -3363,8 +3361,7 @@ fn fill_cell_payload(
         PageType::TableLeaf | PageType::IndexLeaf
     ));
     // TODO: make record raw from start, having to serialize is not good
-    let mut record_buf = Vec::new();
-    record.serialize(&mut record_buf);
+    let record_buf = record.payload.to_vec();
 
     // fill in header
     if matches!(page_type, PageType::TableLeaf) {
@@ -3537,6 +3534,7 @@ mod tests {
     use crate::storage::sqlite3_ondisk;
     use crate::storage::sqlite3_ondisk::DatabaseHeader;
     use crate::types::Text;
+    use crate::vdbe::Register;
     use crate::Connection;
     use crate::{BufferPool, DatabaseStorage, WalFile, WalFileShared, WriteCompletion};
     use std::cell::RefCell;
@@ -3558,7 +3556,7 @@ mod tests {
             pager::PageRef,
             sqlite3_ondisk::{BTreeCell, PageContent, PageType},
         },
-        types::{OwnedValue, Record},
+        types::OwnedValue,
         Database, Page, Pager, PlatformIO,
     };
 
@@ -3616,7 +3614,7 @@ mod tests {
         id: usize,
         pos: usize,
         page: &mut PageContent,
-        record: Record,
+        record: ImmutableRecord,
         conn: &Rc<Connection>,
     ) -> Vec<u8> {
         let mut payload: Vec<u8> = Vec::new();
@@ -3639,7 +3637,8 @@ mod tests {
         let page = get_page(2);
         let page = page.get_contents();
         let header_size = 8;
-        let record = Record::new([OwnedValue::Integer(1)].to_vec());
+        let record =
+            ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Integer(1))]);
         let payload = add_record(1, 0, page, record, &conn);
         assert_eq!(page.cell_count(), 1);
         let free = compute_free_space(page, 4096);
@@ -3667,7 +3666,9 @@ mod tests {
         let mut cells = Vec::new();
         let usable_space = 4096;
         for i in 0..3 {
-            let record = Record::new([OwnedValue::Integer(i as i64)].to_vec());
+            let record = ImmutableRecord::from_registers(&[Register::OwnedValue(
+                OwnedValue::Integer(i as i64),
+            )]);
             let payload = add_record(i, i, page, record, &conn);
             assert_eq!(page.cell_count(), i + 1);
             let free = compute_free_space(page, usable_space);
@@ -3892,7 +3893,9 @@ mod tests {
                 )
                 .unwrap();
                 let key = OwnedValue::Integer(*key);
-                let value = Record::new(vec![OwnedValue::Blob(Rc::new(vec![0; *size]))]);
+                let value = ImmutableRecord::from_registers(&[Register::OwnedValue(
+                    OwnedValue::Blob(Rc::new(vec![0; *size])),
+                )]);
                 tracing::info!("insert key:{}", key);
                 run_until_done(|| cursor.insert(&key, &value, true), pager.deref()).unwrap();
                 tracing::info!(
@@ -3957,7 +3960,9 @@ mod tests {
                 .unwrap();
 
                 let key = OwnedValue::Integer(key);
-                let value = Record::new(vec![OwnedValue::Blob(Rc::new(vec![0; size]))]);
+                let value = ImmutableRecord::from_registers(&[Register::OwnedValue(
+                    OwnedValue::Blob(Rc::new(vec![0; size])),
+                )]);
                 run_until_done(|| cursor.insert(&key, &value, true), pager.deref()).unwrap();
             }
             tracing::info!(
@@ -3994,7 +3999,9 @@ mod tests {
         let usable_space = 4096;
         let total_cells = 10;
         for i in 0..total_cells {
-            let record = Record::new([OwnedValue::Integer(i as i64)].to_vec());
+            let record = ImmutableRecord::from_registers(&[Register::OwnedValue(
+                OwnedValue::Integer(i as i64),
+            )]);
             let payload = add_record(i, i, page, record, &conn);
             assert_eq!(page.cell_count(), i + 1);
             let free = compute_free_space(page, usable_space);
@@ -4352,7 +4359,9 @@ mod tests {
         let mut cells = Vec::new();
         let usable_space = 4096;
         for i in 0..3 {
-            let record = Record::new([OwnedValue::Integer(i as i64)].to_vec());
+            let record = ImmutableRecord::from_registers(&[Register::OwnedValue(
+                OwnedValue::Integer(i as i64),
+            )]);
             let payload = add_record(i, i, page, record, &conn);
             assert_eq!(page.cell_count(), i + 1);
             let free = compute_free_space(page, usable_space);
@@ -4392,7 +4401,9 @@ mod tests {
         let usable_space = 4096;
         let total_cells = 10;
         for i in 0..total_cells {
-            let record = Record::new([OwnedValue::Integer(i as i64)].to_vec());
+            let record = ImmutableRecord::from_registers(&[Register::OwnedValue(
+                OwnedValue::Integer(i as i64),
+            )]);
             let payload = add_record(i, i, page, record, &conn);
             assert_eq!(page.cell_count(), i + 1);
             let free = compute_free_space(page, usable_space);
@@ -4448,7 +4459,9 @@ mod tests {
                     // allow appends with extra place to insert
                     let cell_idx = rng.next_u64() as usize % (page.cell_count() + 1);
                     let free = compute_free_space(page, usable_space);
-                    let record = Record::new([OwnedValue::Integer(i as i64)].to_vec());
+                    let record = ImmutableRecord::from_registers(&[Register::OwnedValue(
+                        OwnedValue::Integer(i as i64),
+                    )]);
                     let mut payload: Vec<u8> = Vec::new();
                     fill_cell_payload(
                         page.page_type(),
@@ -4517,7 +4530,9 @@ mod tests {
                         // allow appends with extra place to insert
                         let cell_idx = rng.next_u64() as usize % (page.cell_count() + 1);
                         let free = compute_free_space(page, usable_space);
-                        let record = Record::new([OwnedValue::Integer(i as i64)].to_vec());
+                        let record = ImmutableRecord::from_registers(&[Register::OwnedValue(
+                            OwnedValue::Integer(i as i64),
+                        )]);
                         let mut payload: Vec<u8> = Vec::new();
                         fill_cell_payload(
                             page.page_type(),
@@ -4571,7 +4586,8 @@ mod tests {
         let header_size = 8;
         let usable_space = 4096;
 
-        let record = Record::new([OwnedValue::Integer(0)].to_vec());
+        let record =
+            ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Integer(0))]);
         let payload = add_record(0, 0, page, record, &conn);
         let free = compute_free_space(page, usable_space);
         assert_eq!(free, 4096 - payload.len() as u16 - 2 - header_size);
@@ -4586,7 +4602,8 @@ mod tests {
         let page = page.get_contents();
         let usable_space = 4096;
 
-        let record = Record::new([OwnedValue::Integer(0)].to_vec());
+        let record =
+            ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Integer(0))]);
         let payload = add_record(0, 0, page, record, &conn);
 
         assert_eq!(page.cell_count(), 1);
@@ -4611,20 +4628,18 @@ mod tests {
         let page = page.get_contents();
         let usable_space = 4096;
 
-        let record = Record::new(
-            [
-                OwnedValue::Integer(0),
-                OwnedValue::Text(Text::new("aaaaaaaa")),
-            ]
-            .to_vec(),
-        );
+        let record = ImmutableRecord::from_registers(&[
+            Register::OwnedValue(OwnedValue::Integer(0)),
+            Register::OwnedValue(OwnedValue::Text(Text::new("aaaaaaaa"))),
+        ]);
         let _ = add_record(0, 0, page, record, &conn);
 
         assert_eq!(page.cell_count(), 1);
         drop_cell(page, 0, usable_space).unwrap();
         assert_eq!(page.cell_count(), 0);
 
-        let record = Record::new([OwnedValue::Integer(0)].to_vec());
+        let record =
+            ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Integer(0))]);
         let payload = add_record(0, 0, page, record, &conn);
         assert_eq!(page.cell_count(), 1);
 
@@ -4647,13 +4662,10 @@ mod tests {
         let page = page.get_contents();
         let usable_space = 4096;
 
-        let record = Record::new(
-            [
-                OwnedValue::Integer(0),
-                OwnedValue::Text(Text::new("aaaaaaaa")),
-            ]
-            .to_vec(),
-        );
+        let record = ImmutableRecord::from_registers(&[
+            Register::OwnedValue(OwnedValue::Integer(0)),
+            Register::OwnedValue(OwnedValue::Text(Text::new("aaaaaaaa"))),
+        ]);
         let _ = add_record(0, 0, page, record, &conn);
 
         for _ in 0..100 {
@@ -4661,7 +4673,8 @@ mod tests {
             drop_cell(page, 0, usable_space).unwrap();
             assert_eq!(page.cell_count(), 0);
 
-            let record = Record::new([OwnedValue::Integer(0)].to_vec());
+            let record =
+                ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Integer(0))]);
             let payload = add_record(0, 0, page, record, &conn);
             assert_eq!(page.cell_count(), 1);
 
@@ -4685,11 +4698,14 @@ mod tests {
         let page = page.get_contents();
         let usable_space = 4096;
 
-        let record = Record::new([OwnedValue::Integer(0)].to_vec());
+        let record =
+            ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Integer(0))]);
         let payload = add_record(0, 0, page, record, &conn);
-        let record = Record::new([OwnedValue::Integer(1)].to_vec());
+        let record =
+            ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Integer(1))]);
         let _ = add_record(1, 1, page, record, &conn);
-        let record = Record::new([OwnedValue::Integer(2)].to_vec());
+        let record =
+            ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Integer(2))]);
         let _ = add_record(2, 2, page, record, &conn);
 
         drop_cell(page, 1, usable_space).unwrap();
@@ -4707,21 +4723,25 @@ mod tests {
         let page = page.get_contents();
         let usable_space = 4096;
 
-        let record = Record::new([OwnedValue::Integer(0)].to_vec());
+        let record =
+            ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Integer(0))]);
         let _ = add_record(0, 0, page, record, &conn);
 
-        let record = Record::new([OwnedValue::Integer(0)].to_vec());
+        let record =
+            ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Integer(0))]);
         let _ = add_record(0, 0, page, record, &conn);
         drop_cell(page, 0, usable_space).unwrap();
 
         defragment_page(page, usable_space);
 
-        let record = Record::new([OwnedValue::Integer(0)].to_vec());
+        let record =
+            ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Integer(0))]);
         let _ = add_record(0, 1, page, record, &conn);
 
         drop_cell(page, 0, usable_space).unwrap();
 
-        let record = Record::new([OwnedValue::Integer(0)].to_vec());
+        let record =
+            ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Integer(0))]);
         let _ = add_record(0, 1, page, record, &conn);
     }
 
@@ -4733,7 +4753,8 @@ mod tests {
         let page = get_page(2);
         let usable_space = 4096;
         let insert = |pos, page| {
-            let record = Record::new([OwnedValue::Integer(0)].to_vec());
+            let record =
+                ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Integer(0))]);
             let _ = add_record(0, pos, page, record, &conn);
         };
         let drop = |pos, page| {
@@ -4772,7 +4793,8 @@ mod tests {
         let page = get_page(2);
         let usable_space = 4096;
         let insert = |pos, page| {
-            let record = Record::new([OwnedValue::Integer(0)].to_vec());
+            let record =
+                ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Integer(0))]);
             let _ = add_record(0, pos, page, record, &conn);
         };
         let drop = |pos, page| {
@@ -4781,7 +4803,8 @@ mod tests {
         let defragment = |page| {
             defragment_page(page, usable_space);
         };
-        let record = Record::new([OwnedValue::Integer(0)].to_vec());
+        let record =
+            ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Integer(0))]);
         let mut payload: Vec<u8> = Vec::new();
         fill_cell_payload(
             page.get_contents().page_type(),
@@ -4815,7 +4838,8 @@ mod tests {
             let mut cursor = BTreeCursor::new(None, pager.clone(), root_page);
             tracing::info!("INSERT INTO t VALUES ({});", i,);
             let key = OwnedValue::Integer(i);
-            let value = Record::new(vec![OwnedValue::Integer(i)]);
+            let value =
+                ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Integer(i))]);
             tracing::trace!("before insert {}", i);
             run_until_done(
                 || {
@@ -4850,7 +4874,9 @@ mod tests {
 
         let page = get_page(2);
         let usable_space = 4096;
-        let record = Record::new([OwnedValue::Blob(Rc::new(vec![0; 3600]))].to_vec());
+        let record = ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Blob(
+            Rc::new(vec![0; 3600]),
+        ))]);
         let mut payload: Vec<u8> = Vec::new();
         fill_cell_payload(
             page.get_contents().page_type(),
@@ -4886,7 +4912,9 @@ mod tests {
         for i in 1..=10000 {
             let mut cursor = BTreeCursor::new(None, pager.clone(), root_page);
             let key = OwnedValue::Integer(i);
-            let value = Record::new(vec![OwnedValue::Text(Text::new("hello world"))]);
+            let value = ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Text(
+                Text::new("hello world"),
+            ))]);
 
             run_until_done(
                 || {
@@ -4957,13 +4985,11 @@ mod tests {
             let mut cursor = BTreeCursor::new(None, pager.clone(), root_page);
             tracing::info!("INSERT INTO t VALUES ({});", i,);
             let key = OwnedValue::Integer(i as i64);
-            let value = Record::new(
-                [OwnedValue::Text(Text {
+            let value =
+                ImmutableRecord::from_registers(&[Register::OwnedValue(OwnedValue::Text(Text {
                     value: Rc::new(huge_texts[i].as_bytes().to_vec()),
                     subtype: crate::types::TextSubtype::Text,
-                })]
-                .to_vec(),
-            );
+                }))]);
             tracing::trace!("before insert {}", i);
             tracing::debug!(
                 "=========== btree before ===========\n{}\n\n",

core/types.rs

@@ -4,9 +4,10 @@ use crate::error::LimboError;
 use crate::ext::{ExtValue, ExtValueType};
 use crate::pseudo::PseudoCursor;
 use crate::storage::btree::BTreeCursor;
+use crate::storage::buffer_pool;
 use crate::storage::sqlite3_ondisk::write_varint;
 use crate::vdbe::sorter::Sorter;
-use crate::vdbe::VTabOpaqueCursor;
+use crate::vdbe::{Register, VTabOpaqueCursor};
 use crate::Result;
 use std::cmp::Ordering;
 use std::fmt::Display;
@@ -71,6 +72,16 @@ impl Text {
     }
 }
 
+impl TextRef {
+    pub fn as_str(&self) -> &str {
+        unsafe { std::str::from_utf8_unchecked(self.value.to_slice()) }
+    }
+
+    pub fn to_string(&self) -> String {
+        self.as_str().to_string()
+    }
+}
+
 #[derive(Debug, Clone)]
 pub enum OwnedValue {
     Null,
@@ -132,6 +143,26 @@ impl OwnedValue {
             OwnedValue::Blob(_) => OwnedValueType::Blob,
         }
     }
+    pub fn serialize_serial(&self, out: &mut Vec<u8>) {
+        match self {
+            OwnedValue::Null => {}
+            OwnedValue::Integer(i) => {
+                let serial_type = SerialType::from(self);
+                match serial_type {
+                    SerialType::I8 => out.extend_from_slice(&(*i as i8).to_be_bytes()),
+                    SerialType::I16 => out.extend_from_slice(&(*i as i16).to_be_bytes()),
+                    SerialType::I24 => out.extend_from_slice(&(*i as i32).to_be_bytes()[1..]), // remove most significant byte
+                    SerialType::I32 => out.extend_from_slice(&(*i as i32).to_be_bytes()),
+                    SerialType::I48 => out.extend_from_slice(&i.to_be_bytes()[2..]), // remove 2 most significant bytes
+                    SerialType::I64 => out.extend_from_slice(&i.to_be_bytes()),
+                    _ => unreachable!(),
+                }
+            }
+            OwnedValue::Float(f) => out.extend_from_slice(&f.to_be_bytes()),
+            OwnedValue::Text(t) => out.extend_from_slice(&t.value),
+            OwnedValue::Blob(b) => out.extend_from_slice(b),
+        };
+    }
 }
 
 #[derive(Debug, Clone, PartialEq)]
@@ -562,33 +593,33 @@ impl std::ops::DivAssign<OwnedValue> for OwnedValue {
 }
 
 pub trait FromValue<'a> {
-    fn from_value(value: &'a OwnedValue) -> Result<Self>
+    fn from_value(value: &'a RefValue) -> Result<Self>
     where
         Self: Sized + 'a;
 }
 
 impl<'a> FromValue<'a> for i64 {
-    fn from_value(value: &'a OwnedValue) -> Result<Self> {
+    fn from_value(value: &'a RefValue) -> Result<Self> {
         match value {
-            OwnedValue::Integer(i) => Ok(*i),
+            RefValue::Integer(i) => Ok(*i),
             _ => Err(LimboError::ConversionError("Expected integer value".into())),
         }
     }
 }
 
 impl<'a> FromValue<'a> for String {
-    fn from_value(value: &'a OwnedValue) -> Result<Self> {
+    fn from_value(value: &'a RefValue) -> Result<Self> {
         match value {
-            OwnedValue::Text(s) => Ok(s.as_str().to_string()),
+            RefValue::Text(s) => Ok(s.as_str().to_string()),
             _ => Err(LimboError::ConversionError("Expected text value".into())),
         }
     }
 }
 
 impl<'a> FromValue<'a> for &'a str {
-    fn from_value(value: &'a OwnedValue) -> Result<Self> {
+    fn from_value(value: &'a RefValue) -> Result<Self> {
         match value {
-            OwnedValue::Text(s) => Ok(s.as_str()),
+            RefValue::Text(s) => Ok(s.as_str()),
             _ => Err(LimboError::ConversionError("Expected text value".into())),
         }
     }
@@ -609,10 +640,10 @@ pub struct Record {
 }
 
 impl Record {
-    pub fn get<'a, T: FromValue<'a> + 'a>(&'a self, idx: usize) -> Result<T> {
-        let value = &self.values[idx];
-        T::from_value(value)
-    }
+    // pub fn get<'a, T: FromValue<'a> + 'a>(&'a self, idx: usize) -> Result<T> {
+    //     let value = &self.values[idx];
+    //     T::from_value(value)
+    // }
 
     pub fn count(&self) -> usize {
         self.values.len()
@@ -636,10 +667,13 @@ impl Record {
 }
 
 impl ImmutableRecord {
-    // pub fn get<'a, T: FromValue<'a> + 'a>(&'a self, idx: usize) -> Result<T> {
-    //     let value = &self.values[idx];
-    //     T::from_value(value)
-    // }
+    pub fn get<'a, T: FromValue<'a> + 'a>(&'a self, idx: usize) -> Result<T> {
+        let value = self
+            .values
+            .get(idx)
+            .ok_or(LimboError::InternalError("Index out of bounds".into()))?;
+        T::from_value(value)
+    }
 
     pub fn count(&self) -> usize {
         self.values.len()
@@ -660,9 +694,166 @@ impl ImmutableRecord {
     pub fn len(&self) -> usize {
         self.values.len()
     }
+
+    pub fn from_registers(registers: &[Register]) -> Self {
+        let mut values = Vec::with_capacity(registers.len());
+        let mut size_header = 0;
+        let mut size_values = 0;
+
+        let mut serial_type_buf = [0; 9];
+        // write serial types
+        for value in registers {
+            let value = value.get_owned_value();
+            let serial_type = SerialType::from(value);
+            let n = write_varint(&mut serial_type_buf[0..], serial_type.into());
+
+            let value_size = match serial_type {
+                SerialType::Null => 0,
+                SerialType::I8 => 8,
+                SerialType::I16 => 16,
+                SerialType::I24 => 24,
+                SerialType::I32 => 32,
+                SerialType::I48 => 48,
+                SerialType::I64 => 64,
+                SerialType::F64 => 64,
+                SerialType::Text { content_size } => content_size,
+                SerialType::Blob { content_size } => content_size,
+            };
+
+            size_header += n;
+            size_values += value_size;
+        }
+        let mut header_size = size_header;
+        let mut header_bytes_buf: Vec<u8> = Vec::new();
+        if header_size <= 126 {
+            // common case
+            header_size += 1;
+        } else {
+            todo!("calculate big header size extra bytes");
+            // get header varint len
+            // header_size += n;
+            // if( nVarint<sqlite3VarintLen(nHdr) ) nHdr++;
+        }
+        // 1. write header size
+        let mut buf = Vec::with_capacity(header_size + size_values);
+        assert!(header_size <= 126);
+        buf.extend(std::iter::repeat(0).take(9));
+        let n = write_varint(buf.as_mut_slice(), header_size as u64);
+        buf.truncate(n);
+        // 2. Write serial
+        for value in registers {
+            let value = value.get_owned_value();
+            let serial_type = SerialType::from(value);
+            let start_pos = buf.len();
+            buf.resize(buf.len() + 9, 0); // Ensure space for varint (1-9 bytes in length)
+            let len = buf.len();
+            let n = write_varint(&mut buf[start_pos..], serial_type.into());
+            buf.truncate(buf.len() - 9 + n); // Remove unused bytes
+        }
+
+        // write content
+        for value in registers {
+            let value = value.get_owned_value();
+            let start_offset = buf.len();
+            match value {
+                OwnedValue::Null => {}
+                OwnedValue::Integer(i) => {
+                    values.push(RefValue::Integer(*i));
+                    let serial_type = SerialType::from(value);
+                    match serial_type {
+                        SerialType::I8 => buf.extend_from_slice(&(*i as i8).to_be_bytes()),
+                        SerialType::I16 => buf.extend_from_slice(&(*i as i16).to_be_bytes()),
+                        SerialType::I24 => buf.extend_from_slice(&(*i as i32).to_be_bytes()[1..]), // remove most significant byte
+                        SerialType::I32 => buf.extend_from_slice(&(*i as i32).to_be_bytes()),
+                        SerialType::I48 => buf.extend_from_slice(&i.to_be_bytes()[2..]), // remove 2 most significant bytes
+                        SerialType::I64 => buf.extend_from_slice(&i.to_be_bytes()),
+                        _ => unreachable!(),
+                    }
+                }
+                OwnedValue::Float(f) => {
+                    values.push(RefValue::Float(*f));
+                    buf.extend_from_slice(&f.to_be_bytes())
+                }
+                OwnedValue::Text(t) => {
+                    buf.extend_from_slice(&t.value);
+                    let end_offset = buf.len();
+                    let len = end_offset - start_offset;
+                    let ptr = unsafe { buf.as_ptr().add(start_offset) };
+                    let value = RefValue::Text(TextRef {
+                        value: RawSlice::new(ptr, len),
+                        subtype: t.subtype.clone(),
+                    });
+                    values.push(value);
+                }
+                OwnedValue::Blob(b) => {
+                    buf.extend_from_slice(b);
+                    let end_offset = buf.len();
+                    let len = end_offset - start_offset;
+                    let ptr = unsafe { buf.as_ptr().add(start_offset) };
+                    values.push(RefValue::Blob(RawSlice::new(ptr, len)));
+                }
+            };
+        }
+
+        Self {
+            payload: Pin::new(buf),
+            values,
+        }
+    }
+}
+
+impl Clone for ImmutableRecord {
+    fn clone(&self) -> Self {
+        let mut new_values = Vec::new();
+        let new_payload = self.payload.clone();
+        for value in &self.values {
+            let value = match value {
+                RefValue::Null => RefValue::Null,
+                RefValue::Integer(i) => RefValue::Integer(*i),
+                RefValue::Float(f) => RefValue::Float(*f),
+                RefValue::Text(text_ref) => {
+                    // let's update pointer
+                    let ptr_start = self.payload.as_ptr() as usize;
+                    let ptr_end = text_ref.value.data as usize;
+                    let len = ptr_end - ptr_start;
+                    let new_ptr = unsafe { new_payload.as_ptr().add(len) };
+                    RefValue::Text(TextRef {
+                        value: RawSlice::new(new_ptr, text_ref.value.len),
+                        subtype: text_ref.subtype.clone(),
+                    })
+                }
+                RefValue::Blob(raw_slice) => {
+                    let ptr_start = self.payload.as_ptr() as usize;
+                    let ptr_end = raw_slice.data as usize;
+                    let len = ptr_end - ptr_start;
+                    let new_ptr = unsafe { new_payload.as_ptr().add(len) };
+                    RefValue::Blob(RawSlice::new(new_ptr, raw_slice.len))
+                }
+            };
+            new_values.push(value);
+        }
+        Self {
+            payload: new_payload,
+            values: new_values,
+        }
+    }
 }
 
 impl RefValue {
+    pub fn to_ffi(&self) -> ExtValue {
+        match self {
+            Self::Null => ExtValue::null(),
+            Self::Integer(i) => ExtValue::from_integer(*i),
+            Self::Float(fl) => ExtValue::from_float(*fl),
+            Self::Text(text) => ExtValue::from_text(
+                std::str::from_utf8(text.value.to_slice())
+                    .unwrap()
+                    .to_string(),
+            ),
+            Self::Blob(blob) => ExtValue::from_blob(blob.to_slice().to_vec()),
+        }
+    }
+
     pub fn to_owned(&self) -> OwnedValue {
         match self {
             RefValue::Null => OwnedValue::Null,
@@ -675,6 +866,24 @@ impl RefValue {
             RefValue::Blob(b) => OwnedValue::Blob(Rc::new(b.to_slice().to_vec())),
         }
     }
+    pub fn to_blob(&self) -> Option<&[u8]> {
+        match self {
+            Self::Blob(blob) => Some(blob.to_slice()),
+            _ => None,
+        }
+    }
+}
+
+impl Display for RefValue {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Null => write!(f, "NULL"),
+            Self::Integer(i) => write!(f, "{}", i),
+            Self::Float(fl) => write!(f, "{:?}", fl),
+            Self::Text(s) => write!(f, "{}", s.as_str()),
+            Self::Blob(b) => write!(f, "{}", String::from_utf8_lossy(b.to_slice())),
+        }
+    }
 }
 impl Eq for RefValue {}
 
@@ -873,6 +1082,18 @@ pub fn compare_immutable_to_record(
     Ordering::Equal
 }
 
+pub fn compare_immutable(l: &[RefValue], r: &[RefValue]) -> std::cmp::Ordering {
+    for (a, b) in l.iter().zip(r.iter()) {
+        match a.partial_cmp(b).unwrap() {
+            Ordering::Equal => {}
+            order => {
+                return order;
+            }
+        }
+    }
+    Ordering::Equal
+}
+
 const I8_LOW: i64 = -128;
 const I8_HIGH: i64 = 127;
 const I16_LOW: i64 = -32768;
@@ -1063,7 +1284,7 @@ pub enum SeekOp {
 #[derive(Clone, PartialEq, Debug)]
 pub enum SeekKey<'a> {
     TableRowId(u64),
-    IndexKey(&'a Record),
+    IndexKey(&'a ImmutableRecord),
 }
 
 impl RawSlice {

core/vdbe/mod.rs

@@ -40,7 +40,8 @@ use crate::storage::wal::CheckpointResult;
 use crate::storage::{btree::BTreeCursor, pager::Pager};
 use crate::translate::plan::{ResultSetColumn, TableReference};
 use crate::types::{
-    AggContext, Cursor, CursorResult, ExternalAggState, OwnedValue, Record, SeekKey, SeekOp,
+    AggContext, Cursor, CursorResult, ExternalAggState, ImmutableRecord, OwnedValue, SeekKey,
+    SeekOp,
 };
 use crate::util::{
     cast_real_to_integer, cast_text_to_integer, cast_text_to_numeric, cast_text_to_real,
@@ -235,7 +236,7 @@ enum HaltState {
 pub enum Register {
     OwnedValue(OwnedValue),
     Aggregate(AggContext),
-    Record(Record),
+    Record(ImmutableRecord),
 }
 
 /// The program state describes the environment in which the program executes.
@@ -243,7 +244,7 @@ pub struct ProgramState {
     pub pc: InsnReference,
     cursors: RefCell<Vec<Option<Cursor>>>,
     registers: Vec<Register>,
-    pub(crate) result_row: Option<Record>,
+    pub(crate) result_row: Option<ImmutableRecord>,
     last_compare: Option<std::cmp::Ordering>,
     deferred_seek: Option<(CursorID, CursorID)>,
     ended_coroutine: Bitfield<4>, // flag to indicate that a coroutine has ended (key is the yield register. currently we assume that the yield register is always between 0-255, YOLO)
@@ -1199,7 +1200,7 @@ impl Program {
                             };
                             if let Some(record) = record {
                                 state.registers[*dest] =
-                                    Register::OwnedValue(record.get_value(*column).clone());
+                                    Register::OwnedValue(record.get_value(*column).to_owned());
                             } else {
                                 state.registers[*dest] = Register::OwnedValue(OwnedValue::Null);
                             }
@@ -1209,7 +1210,7 @@ impl Program {
                                 let mut cursor = state.get_cursor(*cursor_id);
                                 let cursor = cursor.as_pseudo_mut();
                                 if let Some(record) = cursor.record() {
-                                    record.get_value(*column).clone()
+                                    record.get_value(*column).to_owned()
                                 } else {
                                     OwnedValue::Null
                                 }
@@ -1230,12 +1231,12 @@ impl Program {
                     count,
                     dest_reg,
                 } => {
-                    let record = make_owned_record(&state.registers, start_reg, count);
+                    let record = make_record(&state.registers, start_reg, count);
                     state.registers[*dest_reg] = Register::Record(record);
                     state.pc += 1;
                 }
                 Insn::ResultRow { start_reg, count } => {
-                    let record = make_owned_record(&state.registers, start_reg, count);
+                    let record = make_record(&state.registers, start_reg, count);
                     state.result_row = Some(record);
                     state.pc += 1;
                     return Ok(StepResult::Row);
@@ -1547,7 +1548,7 @@ impl Program {
                             let mut cursor = state.get_cursor(*cursor_id);
                             let cursor = cursor.as_btree_mut();
                             let record_from_regs =
-                                make_owned_record(&state.registers, start_reg, num_regs);
+                                make_record(&state.registers, start_reg, num_regs);
                             let found = return_if_io!(
                                 cursor.seek(SeekKey::IndexKey(&record_from_regs), SeekOp::GE)
                             );
@@ -1605,8 +1606,8 @@ impl Program {
                         let found = {
                             let mut cursor = state.get_cursor(*cursor_id);
                             let cursor = cursor.as_btree_mut();
-                            let record_from_regs: Record =
-                                make_owned_record(&state.registers, start_reg, num_regs);
+                            let record_from_regs =
+                                make_record(&state.registers, start_reg, num_regs);
                             let found = return_if_io!(
                                 cursor.seek(SeekKey::IndexKey(&record_from_regs), SeekOp::GT)
                             );
@@ -1663,8 +1664,7 @@ impl Program {
                     let pc = {
                         let mut cursor = state.get_cursor(*cursor_id);
                         let cursor = cursor.as_btree_mut();
-                        let record_from_regs: Record =
-                            make_owned_record(&state.registers, start_reg, num_regs);
+                        let record_from_regs = make_record(&state.registers, start_reg, num_regs);
                         let pc = if let Some(ref idx_record) = *cursor.record() {
                             // Compare against the same number of values
                             if idx_record.get_values()[..record_from_regs.len()]
@@ -1693,8 +1693,7 @@ impl Program {
                     let pc = {
                         let mut cursor = state.get_cursor(*cursor_id);
                         let cursor = cursor.as_btree_mut();
-                        let record_from_regs: Record =
-                            make_owned_record(&state.registers, start_reg, num_regs);
+                        let record_from_regs = make_record(&state.registers, start_reg, num_regs);
                         let pc = if let Some(ref idx_record) = *cursor.record() {
                             // Compare against the same number of values
                             if idx_record.get_values()[..record_from_regs.len()]
@@ -1723,8 +1722,7 @@ impl Program {
                     let pc = {
                         let mut cursor = state.get_cursor(*cursor_id);
                         let cursor = cursor.as_btree_mut();
-                        let record_from_regs: Record =
-                            make_owned_record(&state.registers, start_reg, num_regs);
+                        let record_from_regs = make_record(&state.registers, start_reg, num_regs);
                         let pc = if let Some(ref idx_record) = *cursor.record() {
                             // Compare against the same number of values
                             if idx_record.get_values()[..record_from_regs.len()]
@@ -1753,8 +1751,7 @@ impl Program {
                     let pc = {
                         let mut cursor = state.get_cursor(*cursor_id);
                         let cursor = cursor.as_btree_mut();
-                        let record_from_regs: Record =
-                            make_owned_record(&state.registers, start_reg, num_regs);
+                        let record_from_regs = make_record(&state.registers, start_reg, num_regs);
                         let pc = if let Some(ref idx_record) = *cursor.record() {
                             // Compare against the same number of values
                             if idx_record.get_values()[..record_from_regs.len()]
@@ -3606,12 +3603,8 @@ fn get_new_rowid<R: Rng>(cursor: &mut BTreeCursor, mut rng: R) -> Result<CursorR
     Ok(CursorResult::Ok(rowid.try_into().unwrap()))
 }
 
-fn make_owned_record(registers: &[Register], start_reg: &usize, count: &usize) -> Record {
-    let mut values = Vec::with_capacity(*count);
-    for r in registers.iter().skip(*start_reg).take(*count) {
-        values.push(r.get_owned_value().clone())
-    }
-    Record::new(values)
+fn make_record(registers: &[Register], start_reg: &usize, count: &usize) -> ImmutableRecord {
+    ImmutableRecord::from_registers(&registers[*start_reg..*start_reg + *count])
 }
 
 fn trace_insn(program: &Program, addr: InsnReference, insn: &Insn) {

core/vdbe/sorter.rs

@@ -1,9 +1,9 @@
-use crate::types::Record;
+use crate::types::{ImmutableRecord, Record};
 use std::cmp::Ordering;
 
 pub struct Sorter {
-    records: Vec<Record>,
-    current: Option<Record>,
+    records: Vec<ImmutableRecord>,
+    current: Option<ImmutableRecord>,
     order: Vec<bool>,
 }
 
@@ -51,11 +51,11 @@ impl Sorter {
     pub fn next(&mut self) {
         self.current = self.records.pop();
     }
-    pub fn record(&self) -> Option<&Record> {
+    pub fn record(&self) -> Option<&ImmutableRecord> {
         self.current.as_ref()
     }
 
-    pub fn insert(&mut self, record: &Record) {
-        self.records.push(Record::new(record.get_values().to_vec()));
+    pub fn insert(&mut self, record: &ImmutableRecord) {
+        self.records.push(record.clone());
     }
 }

simulator/generation/plan.rs

@@ -494,13 +494,13 @@ impl Interaction {
                         let mut r = Vec::new();
                         for v in row.get_values() {
                             let v = match v {
-                                limbo_core::OwnedValue::Null => Value::Null,
-                                limbo_core::OwnedValue::Integer(i) => Value::Integer(*i),
-                                limbo_core::OwnedValue::Float(f) => Value::Float(*f),
-                                limbo_core::OwnedValue::Text(t) => {
+                                limbo_core::RefValue::Null => Value::Null,
+                                limbo_core::RefValue::Integer(i) => Value::Integer(*i),
+                                limbo_core::RefValue::Float(f) => Value::Float(*f),
+                                limbo_core::RefValue::Text(t) => {
                                     Value::Text(t.as_str().to_string())
                                 }
-                                limbo_core::OwnedValue::Blob(b) => Value::Blob(b.to_vec()),
+                                limbo_core::RefValue::Blob(b) => Value::Blob(b.to_slice().to_vec()),
                             };
                             r.push(v);
                         }

sqlite3/src/lib.rs

@@ -637,7 +637,7 @@ pub unsafe extern "C" fn sqlite3_column_text(
         None => return std::ptr::null(),
     };
     match row.get_values().get(idx as usize) {
-        Some(limbo_core::OwnedValue::Text(text)) => text.as_str().as_ptr(),
+        Some(limbo_core::RefValue::Text(text)) => text.as_str().as_ptr(),
         _ => std::ptr::null(),
     }
 }

tests/integration/functions/test_function_rowid.rs

@@ -30,7 +30,7 @@ fn test_last_insert_rowid_basic() -> anyhow::Result<()> {
             match rows.step()? {
                 StepResult::Row => {
                     let row = rows.row().unwrap();
-                    if let limbo_core::OwnedValue::Integer(id) = row.get_value(0) {
+                    if let limbo_core::RefValue::Integer(id) = row.get_value(0) {
                         assert_eq!(*id, 1, "First insert should have rowid 1");
                     }
                 }
@@ -66,7 +66,7 @@ fn test_last_insert_rowid_basic() -> anyhow::Result<()> {
             match rows.step()? {
                 StepResult::Row => {
                     let row = rows.row().unwrap();
-                    if let limbo_core::OwnedValue::Integer(id) = row.get_value(0) {
+                    if let limbo_core::RefValue::Integer(id) = row.get_value(0) {
                         last_id = *id;
                     }
                 }
@@ -112,7 +112,7 @@ fn test_integer_primary_key() -> anyhow::Result<()> {
         match select_query.step()? {
             StepResult::Row => {
                 let row = select_query.row().unwrap();
-                if let limbo_core::OwnedValue::Integer(id) = row.get_value(0) {
+                if let limbo_core::RefValue::Integer(id) = row.get_value(0) {
                     rowids.push(*id);
                 }
             }

tests/integration/fuzz/mod.rs

@@ -74,11 +74,15 @@ mod tests {
                 .get_values()
                 .iter()
                 .map(|x| match x {
-                    limbo_core::OwnedValue::Null => rusqlite::types::Value::Null,
-                    limbo_core::OwnedValue::Integer(x) => rusqlite::types::Value::Integer(*x),
-                    limbo_core::OwnedValue::Float(x) => rusqlite::types::Value::Real(*x),
-                    limbo_core::OwnedValue::Text(x) => rusqlite::types::Value::Text(x.to_string()),
-                    limbo_core::OwnedValue::Blob(x) => rusqlite::types::Value::Blob(x.to_vec()),
+                    limbo_core::RefValue::Null => rusqlite::types::Value::Null,
+                    limbo_core::RefValue::Integer(x) => rusqlite::types::Value::Integer(*x),
+                    limbo_core::RefValue::Float(x) => rusqlite::types::Value::Real(*x),
+                    limbo_core::RefValue::Text(x) => {
+                        rusqlite::types::Value::Text(x.as_str().to_string())
+                    }
+                    limbo_core::RefValue::Blob(x) => {
+                        rusqlite::types::Value::Blob(x.to_slice().to_vec())
+                    }
                 })
                 .collect();
             rows.push(row);

tests/integration/query_processing/test_read_path.rs

@@ -63,23 +63,23 @@ fn test_statement_bind() -> anyhow::Result<()> {
         match stmt.step()? {
             StepResult::Row => {
                 let row = stmt.row().unwrap();
-                if let limbo_core::OwnedValue::Text(s) = row.get_value(0) {
+                if let limbo_core::RefValue::Text(s) = row.get_value(0) {
                     assert_eq!(s.as_str(), "hello")
                 }
 
-                if let limbo_core::OwnedValue::Text(s) = row.get_value(1) {
+                if let limbo_core::RefValue::Text(s) = row.get_value(1) {
                     assert_eq!(s.as_str(), "hello")
                 }
 
-                if let limbo_core::OwnedValue::Integer(i) = row.get_value(2) {
+                if let limbo_core::RefValue::Integer(i) = row.get_value(2) {
                     assert_eq!(*i, 42)
                 }
 
-                if let limbo_core::OwnedValue::Blob(v) = row.get_value(3) {
-                    assert_eq!(v.as_ref(), &vec![0x1 as u8, 0x2, 0x3])
+                if let limbo_core::RefValue::Blob(v) = row.get_value(3) {
+                    assert_eq!(v.to_slice(), &vec![0x1 as u8, 0x2, 0x3])
                 }
 
-                if let limbo_core::OwnedValue::Float(f) = row.get_value(4) {
+                if let limbo_core::RefValue::Float(f) = row.get_value(4) {
                     assert_eq!(*f, 0.5)
                 }
             }

tests/integration/query_processing/test_write_path.rs

@@ -1,6 +1,6 @@
 use crate::common::{self, maybe_setup_tracing};
 use crate::common::{compare_string, do_flush, TempDatabase};
-use limbo_core::{Connection, StepResult};
+use limbo_core::{Connection, RefValue, StepResult};
 use log::debug;
 use std::rc::Rc;
 
@@ -47,12 +47,12 @@ fn test_simple_overflow_page() -> anyhow::Result<()> {
                     let first_value = row.get_value(0);
                     let text = row.get_value(1);
                     let id = match first_value {
-                        limbo_core::OwnedValue::Integer(i) => *i as i32,
-                        limbo_core::OwnedValue::Float(f) => *f as i32,
+                        limbo_core::RefValue::Integer(i) => *i as i32,
+                        limbo_core::RefValue::Float(f) => *f as i32,
                         _ => unreachable!(),
                     };
                     let text = match text {
-                        limbo_core::OwnedValue::Text(t) => t.as_str(),
+                        limbo_core::RefValue::Text(t) => t.as_str(),
                         _ => unreachable!(),
                     };
                     assert_eq!(1, id);
@@ -123,12 +123,12 @@ fn test_sequential_overflow_page() -> anyhow::Result<()> {
                     let first_value = row.get_value(0);
                     let text = row.get_value(1);
                     let id = match first_value {
-                        limbo_core::OwnedValue::Integer(i) => *i as i32,
-                        limbo_core::OwnedValue::Float(f) => *f as i32,
+                        limbo_core::RefValue::Integer(i) => *i as i32,
+                        limbo_core::RefValue::Float(f) => *f as i32,
                         _ => unreachable!(),
                     };
                     let text = match text {
-                        limbo_core::OwnedValue::Text(t) => t.as_str(),
+                        limbo_core::RefValue::Text(t) => t.as_str(),
                         _ => unreachable!(),
                     };
                     let huge_text = &huge_texts[current_index];
@@ -194,8 +194,8 @@ fn test_sequential_write() -> anyhow::Result<()> {
                         let row = rows.row().unwrap();
                         let first_value = row.get_values().first().expect("missing id");
                         let id = match first_value {
-                            limbo_core::OwnedValue::Integer(i) => *i as i32,
-                            limbo_core::OwnedValue::Float(f) => *f as i32,
+                            limbo_core::RefValue::Integer(i) => *i as i32,
+                            limbo_core::RefValue::Float(f) => *f as i32,
                             _ => unreachable!(),
                         };
                         assert_eq!(current_read_index, id);
@@ -260,7 +260,7 @@ fn test_regression_multi_row_insert() -> anyhow::Result<()> {
                     let row = rows.row().unwrap();
                     let first_value = row.get_values().first().expect("missing id");
                     let id = match first_value {
-                        limbo_core::OwnedValue::Float(f) => *f as i32,
+                        RefValue::Float(f) => *f as i32,
                         _ => panic!("expected float"),
                     };
                     actual_ids.push(id);
@@ -370,8 +370,8 @@ fn test_wal_checkpoint() -> anyhow::Result<()> {
                     let row = rows.row().unwrap();
                     let first_value = row.get_value(0);
                     let id = match first_value {
-                        limbo_core::OwnedValue::Integer(i) => *i as i32,
-                        limbo_core::OwnedValue::Float(f) => *f as i32,
+                        RefValue::Integer(i) => *i as i32,
+                        RefValue::Float(f) => *f as i32,
                         _ => unreachable!(),
                     };
                     assert_eq!(current_index, id as usize);
@@ -434,7 +434,7 @@ fn test_wal_restart() -> anyhow::Result<()> {
                             let row = rows.row().unwrap();
                             let first_value = row.get_value(0);
                             let count = match first_value {
-                                limbo_core::OwnedValue::Integer(i) => i,
+                                RefValue::Integer(i) => i,
                                 _ => unreachable!(),
                             };
                             debug!("counted {}", count);

tests/integration/wal/test_wal.rs

@@ -1,5 +1,5 @@
 use crate::common::{do_flush, maybe_setup_tracing, TempDatabase};
-use limbo_core::{Connection, LimboError, Result, StepResult};
+use limbo_core::{Connection, LimboError, RefValue, Result, StepResult};
 use std::cell::RefCell;
 use std::ops::Deref;
 use std::rc::Rc;
@@ -83,7 +83,7 @@ fn test_wal_1_writer_1_reader() -> Result<()> {
                             let row = rows.row().unwrap();
                             let first_value = row.get_value(0);
                             let id = match first_value {
-                                limbo_core::OwnedValue::Integer(i) => *i as i32,
+                                RefValue::Integer(i) => *i as i32,
                                 _ => unreachable!(),
                             };
                             assert_eq!(id, i);
@@ -158,7 +158,7 @@ pub(crate) fn execute_and_get_ints(
                 let row = stmt.row().unwrap();
                 for value in row.get_values() {
                     let out = match value {
-                        limbo_core::OwnedValue::Integer(i) => i,
+                        limbo_core::RefValue::Integer(i) => i,
                         _ => {
                             return Err(LimboError::ConversionError(format!(
                                 "cannot convert {value} to int"