Merge 'Add json_each table-valued function (1-arg only)' from Mikaël Francoeur

This adds the [`json_each`](https://sqlite.org/json1.html#the_json_each_
and_json_tree_table_valued_functions) TVF. Only the 1-arg version is
supported for now.
As suggested in the comments on this PR, I've also extended the virtual
table system to support internal TVF's, as opposed to extensions and
pragma TVF's.

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

Closes #2691

COMPAT.md

@@ -393,7 +393,7 @@ Modifiers:
 | jsonb_group_array(value)           | Yes     |                                                                                                                                              |
 | json_group_object(label,value)     | Yes     |                                                                                                                                              |
 | jsonb_group_object(name,value)     | Yes     |                                                                                                                                              |
-| json_each(json)                    |         |                                                                                                                                              |
+| json_each(json)                    | Yes     |                                                                                                                                              |
 | json_each(json,path)               |         |                                                                                                                                              |
 | json_tree(json)                    |         |                                                                                                                                              |
 | json_tree(json,path)               |         |                                                                                                                                              |

core/json/jsonb.rs

@@ -841,6 +841,18 @@ impl JsonbHeader {
     }
 }
 
+pub struct ArrayIteratorState {
+    cursor: usize,
+    end: usize,
+    index: usize,
+}
+
+pub struct ObjectIteratorState {
+    cursor: usize,
+    end: usize,
+    index: usize,
+}
+
 impl Jsonb {
     pub fn new(capacity: usize, data: Option<&[u8]>) -> Self {
         if let Some(data) = data {
@@ -2872,6 +2884,94 @@ impl Jsonb {
 
         Ok(())
     }
+
+    pub fn array_iterator(&self) -> Result<ArrayIteratorState> {
+        let (hdr, off) = self.read_header(0)?;
+        match hdr {
+            JsonbHeader(ElementType::ARRAY, len) => Ok(ArrayIteratorState {
+                cursor: off,
+                end: off + len,
+                index: 0,
+            }),
+            _ => bail_parse_error!("jsonb.array_iterator(): not an array"),
+        }
+    }
+
+    pub fn array_iterator_next(
+        &self,
+        st: &ArrayIteratorState,
+    ) -> Option<((usize, Jsonb), ArrayIteratorState)> {
+        if st.cursor >= st.end {
+            return None;
+        }
+
+        let (JsonbHeader(_, payload_len), header_len) = self.read_header(st.cursor).ok()?;
+        let start = st.cursor;
+        let stop = start.checked_add(header_len + payload_len)?;
+
+        if stop > st.end || stop > self.data.len() {
+            return None;
+        }
+
+        let elem = Jsonb::new(stop - start, Some(&self.data[start..stop]));
+        let next = ArrayIteratorState {
+            cursor: stop,
+            end: st.end,
+            index: st.index + 1,
+        };
+
+        Some(((st.index, elem), next))
+    }
+
+    pub fn object_iterator(&self) -> Result<ObjectIteratorState> {
+        let (hdr, off) = self.read_header(0)?;
+        match hdr {
+            JsonbHeader(ElementType::OBJECT, len) => Ok(ObjectIteratorState {
+                cursor: off,
+                end: off + len,
+                index: 0,
+            }),
+            _ => bail_parse_error!("jsonb.object_iterator(): not an object"),
+        }
+    }
+
+    pub fn object_iterator_next(
+        &self,
+        st: &ObjectIteratorState,
+    ) -> Option<((usize, Jsonb, Jsonb), ObjectIteratorState)> {
+        if st.cursor >= st.end {
+            return None;
+        }
+
+        // key
+        let (JsonbHeader(key_ty, key_len), key_hdr_len) = self.read_header(st.cursor).ok()?;
+        if !key_ty.is_valid_key() {
+            return None;
+        }
+        let key_start = st.cursor;
+        let key_stop = key_start.checked_add(key_hdr_len + key_len)?;
+        if key_stop > st.end || key_stop > self.data.len() {
+            return None;
+        }
+
+        // value
+        let (JsonbHeader(_, val_len), val_hdr_len) = self.read_header(key_stop).ok()?;
+        let val_start = key_stop;
+        let val_stop = val_start.checked_add(val_hdr_len + val_len)?;
+        if val_stop > st.end || val_stop > self.data.len() {
+            return None;
+        }
+
+        let key = Jsonb::new(key_stop - key_start, Some(&self.data[key_start..key_stop]));
+        let value = Jsonb::new(val_stop - val_start, Some(&self.data[val_start..val_stop]));
+        let next = ObjectIteratorState {
+            cursor: val_stop,
+            end: st.end,
+            index: st.index + 1,
+        };
+
+        Some(((st.index, key, value), next))
+    }
 }
 
 impl std::str::FromStr for Jsonb {

core/json/mod.rs

@@ -3,6 +3,7 @@ mod error;
 pub(crate) mod jsonb;
 mod ops;
 pub(crate) mod path;
+pub(crate) mod vtab;
 
 use crate::json::error::Error as JsonError;
 pub use crate::json::ops::{

core/json/vtab.rs

@@ -0,0 +1,436 @@
+use std::{cell::RefCell, result::Result, sync::Arc};
+
+use turso_ext::{ConstraintUsage, ResultCode};
+
+use crate::{
+    json::{
+        convert_dbtype_to_jsonb,
+        jsonb::{ArrayIteratorState, Jsonb, ObjectIteratorState},
+        vtab::columns::Columns,
+        Conv,
+    },
+    types::Text,
+    vtab::{InternalVirtualTable, InternalVirtualTableCursor},
+    Connection, LimboError, Value,
+};
+
+use super::jsonb;
+
+pub struct JsonEachVirtualTable;
+
+const COL_KEY: usize = 0;
+const COL_VALUE: usize = 1;
+const COL_TYPE: usize = 2;
+const COL_ATOM: usize = 3;
+const COL_ID: usize = 4;
+const COL_PARENT: usize = 5;
+const COL_FULLKEY: usize = 6;
+const COL_PATH: usize = 7;
+const COL_JSON: usize = 8;
+const COL_ROOT: usize = 9;
+
+impl InternalVirtualTable for JsonEachVirtualTable {
+    fn name(&self) -> String {
+        "json_each".to_owned()
+    }
+
+    fn open(
+        &self,
+        _conn: Arc<Connection>,
+    ) -> crate::Result<std::sync::Arc<RefCell<(dyn InternalVirtualTableCursor + 'static)>>> {
+        Ok(Arc::new(RefCell::new(JsonEachCursor::default())))
+    }
+
+    fn best_index(
+        &self,
+        constraints: &[turso_ext::ConstraintInfo],
+        _order_by: &[turso_ext::OrderByInfo],
+    ) -> Result<turso_ext::IndexInfo, ResultCode> {
+        use turso_ext::ConstraintOp;
+
+        let mut usages = vec![
+            ConstraintUsage {
+                argv_index: None,
+                omit: false
+            };
+            constraints.len()
+        ];
+        let mut have_json = false;
+
+        for (i, c) in constraints.iter().enumerate() {
+            if c.usable && c.op == ConstraintOp::Eq && c.column_index as usize == COL_JSON {
+                usages[i] = ConstraintUsage {
+                    argv_index: Some(1),
+                    omit: true,
+                };
+                have_json = true;
+                break;
+            }
+        }
+
+        Ok(turso_ext::IndexInfo {
+            idx_num: i32::from(have_json),
+            idx_str: None,
+            order_by_consumed: false,
+            estimated_cost: if have_json { 10.0 } else { 1_000_000.0 },
+            estimated_rows: if have_json { 100 } else { u32::MAX },
+            constraint_usages: usages,
+        })
+    }
+
+    fn sql(&self) -> String {
+        "CREATE TABLE json_each(
+            key ANY,             -- key for current element relative to its parent
+            value ANY,           -- value for the current element
+            type TEXT,           -- 'object','array','string','integer', etc.
+            atom ANY,            -- value for primitive types, null for array & object
+            id INTEGER,          -- integer ID for this element
+            parent INTEGER,      -- integer ID for the parent of this element
+            fullkey TEXT,        -- full path describing the current element
+            path TEXT,           -- path to the container of the current row
+            json JSON HIDDEN,    -- 1st input parameter: the raw JSON
+            root TEXT HIDDEN     -- 2nd input parameter: the PATH at which to start
+        );"
+        .to_owned()
+    }
+}
+
+impl std::fmt::Debug for JsonEachVirtualTable {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("JsonEachVirtualTable").finish()
+    }
+}
+
+enum IteratorState {
+    Array(ArrayIteratorState),
+    Object(ObjectIteratorState),
+    Primitive,
+    None,
+}
+
+pub struct JsonEachCursor {
+    rowid: i64,
+    no_more_rows: bool,
+    json: Jsonb,
+    iterator_state: IteratorState,
+    columns: Columns,
+}
+
+impl Default for JsonEachCursor {
+    fn default() -> Self {
+        Self {
+            rowid: 0,
+            no_more_rows: false,
+            json: Jsonb::new(0, None),
+            iterator_state: IteratorState::None,
+            columns: Columns::default(),
+        }
+    }
+}
+
+impl InternalVirtualTableCursor for JsonEachCursor {
+    fn filter(
+        &mut self,
+        args: &[Value],
+        _idx_str: Option<String>,
+        _idx_num: i32,
+    ) -> Result<bool, LimboError> {
+        if args.is_empty() {
+            return Ok(false);
+        }
+        if args.len() == 2 {
+            return Err(LimboError::InvalidArgument(
+                "2-arg json_each is not supported yet".to_owned(),
+            ));
+        }
+        if args.len() != 1 && args.len() != 2 {
+            return Err(LimboError::InvalidArgument(
+                "json_each accepts 1 or 2 arguments".to_owned(),
+            ));
+        }
+
+        let db_value = &args[0];
+
+        let jsonb = convert_dbtype_to_jsonb(db_value, Conv::Strict)?;
+
+        let element_type = jsonb.element_type()?;
+        self.json = jsonb;
+
+        match element_type {
+            jsonb::ElementType::ARRAY => {
+                let iter = self.json.array_iterator()?;
+                self.iterator_state = IteratorState::Array(iter);
+            }
+            jsonb::ElementType::OBJECT => {
+                let iter = self.json.object_iterator()?;
+                self.iterator_state = IteratorState::Object(iter);
+            }
+            jsonb::ElementType::NULL
+            | jsonb::ElementType::TRUE
+            | jsonb::ElementType::FALSE
+            | jsonb::ElementType::INT
+            | jsonb::ElementType::INT5
+            | jsonb::ElementType::FLOAT
+            | jsonb::ElementType::FLOAT5
+            | jsonb::ElementType::TEXT
+            | jsonb::ElementType::TEXT5
+            | jsonb::ElementType::TEXTJ
+            | jsonb::ElementType::TEXTRAW => {
+                self.iterator_state = IteratorState::Primitive;
+            }
+            jsonb::ElementType::RESERVED1
+            | jsonb::ElementType::RESERVED2
+            | jsonb::ElementType::RESERVED3 => {
+                unreachable!("element type not supported: {element_type:?}");
+            }
+        };
+
+        self.next()
+    }
+
+    fn next(&mut self) -> Result<bool, LimboError> {
+        self.rowid += 1;
+        if self.no_more_rows {
+            return Ok(false);
+        }
+
+        match &self.iterator_state {
+            IteratorState::Array(state) => {
+                let Some(((idx, jsonb), new_state)) = self.json.array_iterator_next(state) else {
+                    self.no_more_rows = true;
+                    return Ok(false);
+                };
+                self.iterator_state = IteratorState::Array(new_state);
+                self.columns = Columns::new(columns::Key::Integer(idx as i64), jsonb);
+            }
+            IteratorState::Object(state) => {
+                let Some(((_idx, key, value), new_state)): Option<(
+                    (usize, Jsonb, Jsonb),
+                    ObjectIteratorState,
+                )> = self.json.object_iterator_next(state) else {
+                    self.no_more_rows = true;
+                    return Ok(false);
+                };
+
+                self.iterator_state = IteratorState::Object(new_state);
+                let key = key.to_string();
+                self.columns = Columns::new(columns::Key::String(key), value);
+            }
+            IteratorState::Primitive => {
+                let json = std::mem::replace(&mut self.json, Jsonb::new(0, None));
+                self.columns = Columns::new_from_primitive(json);
+                self.no_more_rows = true;
+            }
+            IteratorState::None => unreachable!(),
+        };
+
+        Ok(true)
+    }
+
+    fn rowid(&self) -> i64 {
+        self.rowid
+    }
+
+    fn column(&self, idx: usize) -> Result<Value, LimboError> {
+        Ok(match idx {
+            COL_KEY => self.columns.key(),
+            COL_VALUE => self.columns.value()?,
+            COL_TYPE => self.columns.ttype(),
+            COL_ATOM => self.columns.atom()?,
+            COL_ID => Value::Integer(self.rowid),
+            COL_PARENT => self.columns.parent(),
+            COL_FULLKEY => self.columns.fullkey(),
+            COL_PATH => self.columns.path(),
+            COL_ROOT => Value::Text(Text::new("json, todo")),
+            _ => Value::Null,
+        })
+    }
+}
+
+mod columns {
+    use crate::{
+        json::{
+            json_string_to_db_type,
+            jsonb::{self, ElementType, Jsonb},
+            OutputVariant,
+        },
+        types::Text,
+        LimboError, Value,
+    };
+
+    #[derive(Debug)]
+    pub(super) enum Key {
+        Integer(i64),
+        String(String),
+    }
+
+    impl Key {
+        fn empty() -> Self {
+            Self::Integer(0)
+        }
+
+        fn fullkey_representation(&self) -> Value {
+            match self {
+                Key::Integer(ref i) => Value::Text(Text::new(&format!("$[{i}]"))),
+                Key::String(ref text) => {
+                    let mut needs_quoting: bool = false;
+
+                    let mut text = (text[1..text.len() - 1]).to_owned();
+                    if text.contains('.') || text.contains(" ") || text.contains('"') {
+                        needs_quoting = true;
+                    }
+
+                    if needs_quoting {
+                        text = format!("\"{text}\"");
+                    }
+                    let s = format!("$.{text}");
+
+                    Value::Text(Text::new(&s))
+                }
+            }
+        }
+
+        fn key_representation(&self) -> Value {
+            match self {
+                Key::Integer(ref i) => Value::Integer(*i),
+                Key::String(ref s) => Value::Text(Text::new(
+                    &s[1..s.len() - 1].to_owned().replace("\\\"", "\""),
+                )),
+            }
+        }
+    }
+
+    pub(super) struct Columns {
+        key: Key,
+        value: Jsonb,
+        is_primitive: bool,
+    }
+
+    impl Default for Columns {
+        fn default() -> Columns {
+            Self {
+                key: Key::empty(),
+                value: Jsonb::new(0, None),
+                is_primitive: false,
+            }
+        }
+    }
+
+    impl Columns {
+        pub(super) fn new(key: Key, value: Jsonb) -> Self {
+            Self {
+                key,
+                value,
+                is_primitive: false,
+            }
+        }
+
+        pub(super) fn new_from_primitive(value: Jsonb) -> Self {
+            Self {
+                key: Key::empty(),
+                value,
+                is_primitive: true,
+            }
+        }
+
+        pub(super) fn atom(&self) -> Result<Value, LimboError> {
+            Self::atom_from_value(&self.value)
+        }
+
+        pub(super) fn value(&self) -> Result<Value, LimboError> {
+            let element_type = self.value.element_type()?;
+            Ok(match element_type {
+                ElementType::ARRAY | ElementType::OBJECT => {
+                    json_string_to_db_type(self.value.clone(), element_type, OutputVariant::String)?
+                }
+                _ => Self::atom_from_value(&self.value)?,
+            })
+        }
+
+        pub(super) fn key(&self) -> Value {
+            if self.is_primitive {
+                return Value::Null;
+            }
+            self.key.key_representation()
+        }
+
+        fn atom_from_value(value: &Jsonb) -> Result<Value, LimboError> {
+            let element_type = value.element_type().expect("invalid value");
+            let string: Result<Value, LimboError> = match element_type {
+                jsonb::ElementType::NULL => Ok(Value::Null),
+                jsonb::ElementType::TRUE => Ok(Value::Integer(1)),
+                jsonb::ElementType::FALSE => Ok(Value::Integer(0)),
+                jsonb::ElementType::INT | jsonb::ElementType::INT5 => Self::jsonb_to_integer(value),
+                jsonb::ElementType::FLOAT | jsonb::ElementType::FLOAT5 => {
+                    Self::jsonb_to_float(value)
+                }
+                jsonb::ElementType::TEXT
+                | jsonb::ElementType::TEXTJ
+                | jsonb::ElementType::TEXT5
+                | jsonb::ElementType::TEXTRAW => {
+                    let s = value.to_string();
+                    let s = (s[1..s.len() - 1]).to_string();
+                    Ok(Value::Text(Text::new(&s)))
+                }
+                jsonb::ElementType::ARRAY => Ok(Value::Null),
+                jsonb::ElementType::OBJECT => Ok(Value::Null),
+                jsonb::ElementType::RESERVED1 => Ok(Value::Null),
+                jsonb::ElementType::RESERVED2 => Ok(Value::Null),
+                jsonb::ElementType::RESERVED3 => Ok(Value::Null),
+            };
+
+            string
+        }
+
+        fn jsonb_to_integer(value: &Jsonb) -> Result<Value, LimboError> {
+            let string = value.to_string();
+            let int = string.parse::<i64>()?;
+
+            Ok(Value::Integer(int))
+        }
+
+        fn jsonb_to_float(value: &Jsonb) -> Result<Value, LimboError> {
+            let string = value.to_string();
+            let float = string.parse::<f64>()?;
+
+            Ok(Value::Float(float))
+        }
+
+        pub(super) fn fullkey(&self) -> Value {
+            if self.is_primitive {
+                return Value::Text(Text::new("$"));
+            }
+            self.key.fullkey_representation()
+        }
+
+        pub(super) fn path(&self) -> Value {
+            Value::Text(Text::new("$"))
+        }
+
+        pub(super) fn parent(&self) -> Value {
+            Value::Null
+        }
+
+        pub(super) fn ttype(&self) -> Value {
+            let element_type = self.value.element_type().expect("invalid value");
+            let ttype = match element_type {
+                jsonb::ElementType::NULL => "null",
+                jsonb::ElementType::TRUE => "true",
+                jsonb::ElementType::FALSE => "false",
+                jsonb::ElementType::INT | jsonb::ElementType::INT5 => "integer",
+                jsonb::ElementType::FLOAT | jsonb::ElementType::FLOAT5 => "real",
+                jsonb::ElementType::TEXT
+                | jsonb::ElementType::TEXTJ
+                | jsonb::ElementType::TEXT5
+                | jsonb::ElementType::TEXTRAW => "text",
+                jsonb::ElementType::ARRAY => "array",
+                jsonb::ElementType::OBJECT => "object",
+                jsonb::ElementType::RESERVED1
+                | jsonb::ElementType::RESERVED2
+                | jsonb::ElementType::RESERVED3 => unreachable!(),
+            };
+
+            Value::Text(Text::new(ttype))
+        }
+    }
+}

core/vtab.rs

@@ -1,8 +1,9 @@
+use crate::json::vtab::JsonEachVirtualTable;
 use crate::pragma::{PragmaVirtualTable, PragmaVirtualTableCursor};
 use crate::schema::Column;
 use crate::util::columns_from_create_table_body;
 use crate::{Connection, LimboError, SymbolTable, Value};
-
+use std::cell::RefCell;
 use std::ffi::c_void;
 use std::ptr::NonNull;
 use std::rc::Rc;
@@ -14,6 +15,7 @@ use turso_parser::{ast, parser::Parser};
 pub(crate) enum VirtualTableType {
     Pragma(PragmaVirtualTable),
     External(ExtVirtualTable),
+    Internal(Arc<RefCell<dyn InternalVirtualTable>>),
 }
 
 #[derive(Clone, Debug)]
@@ -29,23 +31,44 @@ impl VirtualTable {
         match &self.vtab_type {
             VirtualTableType::Pragma(_) => true,
             VirtualTableType::External(table) => table.readonly(),
+            VirtualTableType::Internal(_) => true,
         }
     }
 
     pub(crate) fn builtin_functions() -> Vec<Arc<VirtualTable>> {
-        PragmaVirtualTable::functions()
+        let mut vtables: Vec<Arc<VirtualTable>> = PragmaVirtualTable::functions()
             .into_iter()
             .map(|(tab, schema)| {
                 let vtab = VirtualTable {
                     name: format!("pragma_{}", tab.pragma_name),
                     columns: Self::resolve_columns(schema)
-                        .expect("built-in function schema resolution should not fail"),
+                        .expect("pragma table-valued function schema resolution should not fail"),
                     kind: VTabKind::TableValuedFunction,
                     vtab_type: VirtualTableType::Pragma(tab),
                 };
                 Arc::new(vtab)
             })
-            .collect()
+            .collect();
+
+        #[cfg(feature = "json")]
+        vtables.extend(Self::json_virtual_tables());
+
+        vtables
+    }
+
+    #[cfg(feature = "json")]
+    fn json_virtual_tables() -> Vec<Arc<VirtualTable>> {
+        let json_each = JsonEachVirtualTable {};
+
+        let json_each_virtual_table = VirtualTable {
+            name: json_each.name(),
+            columns: Self::resolve_columns(json_each.sql())
+                .expect("internal table-valued function schema resolution should not fail"),
+            kind: VTabKind::TableValuedFunction,
+            vtab_type: VirtualTableType::Internal(Arc::new(RefCell::new(json_each))),
+        };
+
+        vec![Arc::new(json_each_virtual_table)]
     }
 
     pub(crate) fn function(name: &str, syms: &SymbolTable) -> crate::Result<Arc<VirtualTable>> {
@@ -107,6 +130,9 @@ impl VirtualTable {
             VirtualTableType::External(table) => {
                 Ok(VirtualTableCursor::External(table.open(conn.clone())?))
             }
+            VirtualTableType::Internal(table) => {
+                Ok(VirtualTableCursor::Internal(table.borrow().open(conn)?))
+            }
         }
     }
 
@@ -114,6 +140,7 @@ impl VirtualTable {
         match &self.vtab_type {
             VirtualTableType::Pragma(_) => Err(LimboError::ReadOnly),
             VirtualTableType::External(table) => table.update(args),
+            VirtualTableType::Internal(_) => Err(LimboError::ReadOnly),
         }
     }
 
@@ -121,6 +148,7 @@ impl VirtualTable {
         match &self.vtab_type {
             VirtualTableType::Pragma(_) => Ok(()),
             VirtualTableType::External(table) => table.destroy(),
+            VirtualTableType::Internal(_) => Ok(()),
         }
     }
 
@@ -132,6 +160,7 @@ impl VirtualTable {
         match &self.vtab_type {
             VirtualTableType::Pragma(table) => table.best_index(constraints),
             VirtualTableType::External(table) => table.best_index(constraints, order_by),
+            VirtualTableType::Internal(table) => table.borrow().best_index(constraints, order_by),
         }
     }
 }
@@ -139,6 +168,7 @@ impl VirtualTable {
 pub enum VirtualTableCursor {
     Pragma(Box<PragmaVirtualTableCursor>),
     External(ExtVirtualTableCursor),
+    Internal(Arc<RefCell<dyn InternalVirtualTableCursor>>),
 }
 
 impl VirtualTableCursor {
@@ -146,6 +176,7 @@ impl VirtualTableCursor {
         match self {
             VirtualTableCursor::Pragma(cursor) => cursor.next(),
             VirtualTableCursor::External(cursor) => cursor.next(),
+            VirtualTableCursor::Internal(cursor) => cursor.borrow_mut().next(),
         }
     }
 
@@ -153,6 +184,7 @@ impl VirtualTableCursor {
         match self {
             VirtualTableCursor::Pragma(cursor) => cursor.rowid(),
             VirtualTableCursor::External(cursor) => cursor.rowid(),
+            VirtualTableCursor::Internal(cursor) => cursor.borrow().rowid(),
         }
     }
 
@@ -160,6 +192,7 @@ impl VirtualTableCursor {
         match self {
             VirtualTableCursor::Pragma(cursor) => cursor.column(column),
             VirtualTableCursor::External(cursor) => cursor.column(column),
+            VirtualTableCursor::Internal(cursor) => cursor.borrow().column(column),
         }
     }
 
@@ -175,6 +208,9 @@ impl VirtualTableCursor {
             VirtualTableCursor::External(cursor) => {
                 cursor.filter(idx_num, idx_str, arg_count, args)
             }
+            VirtualTableCursor::Internal(cursor) => {
+                cursor.borrow_mut().filter(&args, idx_str, idx_num)
+            }
         }
     }
 }
@@ -376,3 +412,31 @@ impl Drop for ExtVirtualTableCursor {
         }
     }
 }
+
+pub trait InternalVirtualTable: std::fmt::Debug {
+    fn name(&self) -> String;
+    fn open(
+        &self,
+        conn: Arc<Connection>,
+    ) -> crate::Result<Arc<RefCell<dyn InternalVirtualTableCursor>>>;
+    /// best_index is used by the optimizer. See the comment on `Table::best_index`.
+    fn best_index(
+        &self,
+        constraints: &[turso_ext::ConstraintInfo],
+        order_by: &[turso_ext::OrderByInfo],
+    ) -> Result<turso_ext::IndexInfo, ResultCode>;
+    fn sql(&self) -> String;
+}
+
+pub trait InternalVirtualTableCursor {
+    /// next returns `Ok(true)` if there are more rows, and `Ok(false)` otherwise.
+    fn next(&mut self) -> Result<bool, LimboError>;
+    fn rowid(&self) -> i64;
+    fn column(&self, column: usize) -> Result<Value, LimboError>;
+    fn filter(
+        &mut self,
+        args: &[Value],
+        idx_str: Option<String>,
+        idx_num: i32,
+    ) -> Result<bool, LimboError>;
+}

testing/json.test

@@ -1223,3 +1223,122 @@ do_execsql_test json_remove_with_arrow {
 #   WITH RECURSIVE c(x) AS (VALUES(1) UNION ALL SELECT x+1 FROM c WHERE x<0x1f)
 #   SELECT sum(json_valid(json_quote('a'||char(x)||'z'))) FROM c ORDER BY x;
 # } {31}
+
+do_execsql_test json_each_arrays_heterogeneous_primitives {
+  SELECT key, atom, type, fullkey, path, typeof(key) AS ktype
+  FROM json_each('[1, 2.5, "x", true, false, null]')
+  ORDER BY key;
+} {
+0|1|integer|$[0]|$|integer
+1|2.5|real|$[1]|$|integer
+2|x|text|$[2]|$|integer
+3|1|true|$[3]|$|integer
+4|0|false|$[4]|$|integer
+5||null|$[5]|$|integer
+}
+
+do_execsql_test json_each_arrays_parent_is_always_null {
+  SELECT COUNT(*) FROM json_each('[0,1,2]') WHERE parent IS NOT NULL;
+} {0}
+
+do_execsql_test json_each_arrays_id_uniqueness {
+  SELECT COUNT(*), COUNT(DISTINCT id)
+  FROM json_each('[10,20,30,40]');
+} {4|4}
+
+do_execsql_test json_each_arrays_empty_container_yields_zero_rows {
+  SELECT COUNT(*) FROM json_each('[]');
+} {0}
+
+do_execsql_test json_each_objects_simple_integer_values {
+  SELECT key, atom, type, fullkey, path, typeof(key) AS ktype
+  FROM json_each('{"a":1,"b":2}')
+  ORDER BY key;
+} {
+  {a|1|integer|$.a|$|text}
+  {b|2|integer|$.b|$|text}
+}
+
+do_execsql_test json_each_objects_nested_containers_value_is_valid_json {
+  SELECT key, type, json_valid(value) AS is_json, fullkey, path
+  FROM json_each('{"o":{"x":5},"a":[7,8]}')
+  ORDER BY key;
+} {
+  {a|array|1|$.a|$}
+  {o|object|1|$.o|$}
+}
+
+do_execsql_test json_each_objects_empty_container_yields_zero_rows {
+  SELECT COUNT(*) FROM json_each('{}');
+} {0}
+
+do_execsql_test json_each_objects_keys_require_quoting_in_json_path {
+  SELECT key, fullkey
+  FROM json_each('{"a space":1,"a.b":2,"\"q\"":3}')
+  ORDER BY key DESC;
+} {
+{a.b|$."a.b"}
+{a space|$."a space"}
+{"q"|$."\"q\""}
+}
+
+do_execsql_test json_each_top_level_integer_single_row_key_null {
+  SELECT (key IS NULL), fullkey, path, atom, type
+  FROM json_each('42');
+} {1|$|$|42|integer}
+
+do_execsql_test json_each_top_level_true_single_row_key_null {
+  SELECT (key IS NULL), fullkey, path, atom, type
+  FROM json_each('true');
+} {1|$|$|1|true}
+
+do_execsql_test json_each_top_level_null_single_row_key_null {
+  SELECT (key IS NULL), fullkey, path, (atom IS NULL), type
+  FROM json_each('null');
+} {1|$|$|1|null}
+
+do_execsql_test json_each_atom_equals_value_for_primitives_containers_are_json_text {
+  WITH t AS (
+    SELECT * FROM json_each('[1,"x",{"y":2},[3]]')
+  )
+  SELECT 
+    SUM(type IN ('object','array') AND json_valid(value)=1),
+    SUM(type NOT IN ('object','array') AND value=atom)
+  FROM t;
+} {2|2}
+
+do_execsql_test json_each_typeof_key_array_indices_integer {
+  SELECT GROUP_CONCAT(ktype,'|') FROM (
+    SELECT typeof(key) AS ktype FROM json_each('[0,1]') ORDER BY key
+  );
+} {integer|integer}
+
+do_execsql_test json_each_typeof_key_object_keys_text {
+  SELECT GROUP_CONCAT(ktype,'|') FROM (
+    SELECT typeof(key) AS ktype FROM json_each('{"0":0,"1":1}') ORDER BY key
+  );
+} {text|text}
+
+do_execsql_test json_each_parent_column_always_null {
+  SELECT COUNT(*) FROM json_each('{"a":[1,2,3],"b":{}}') WHERE parent IS NOT NULL;
+} {0}
+
+do_execsql_test_error json_each_malformed_json_raises_error {
+  SELECT * FROM json_each('{not json}');
+} {(.*malformed JSON.*)}
+
+do_execsql_test json_each_object_member_order_preserved {
+  SELECT key FROM json_each('{"z":0,"a":1,"m":2}');
+} {z a m}
+
+do_execsql_test json_each_json_extract_on_value {
+  SELECT key, json_extract(value, '$.x')
+  FROM json_each('{"k1":{"x":11},"k2":{"x":22},"k3":{"x":[3]}}')
+  WHERE type!='array'
+  ORDER BY key;
+} {
+  {k1|11}
+  {k2|22}
+  {k3|[3]}
+}
+