aljaz/turso
1
0
Fork 0
mirror of https://github.com/aljazceru/turso.git synced 2026-02-23 17:05:36 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge 'CREATE TABLE support' from Pekka Enberg

Browse Source 
Closes #383
This commit is contained in:
Pere Diaz Bou
2024-11-18 13:40:20 +01:00
parent 5efc218e6e 80f60852df
commit 9c4ca28bd9
10 changed files with 431 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
core/lib.rs
View File

@@ -29,6 +29,7 @@ use storage::database::FileStorage;
use storage::pager::allocate_page;
use storage::sqlite3_ondisk::{DatabaseHeader, DATABASE_HEADER_SIZE};
pub use storage::wal::WalFile;
use util::parse_schema_rows;
use translate::optimizer::optimize_plan;
use translate::planner::prepare_select_plan;
@@ -59,7 +60,7 @@ enum TransactionState {
pub struct Database {
pager: Rc<Pager>,
schema: Rc<Schema>,
schema: Rc<RefCell<Schema>>,
header: Rc<RefCell<DatabaseHeader>>,
transaction_state: RefCell<TransactionState>,
}
@@ -98,7 +99,7 @@ impl Database {
wal,
io.clone(),
)?);
let bootstrap_schema = Rc::new(Schema::new());
let bootstrap_schema = Rc::new(RefCell::new(Schema::new()));
let conn = Rc::new(Connection {
pager: pager.clone(),
schema: bootstrap_schema.clone(),
@@ -107,46 +108,8 @@ impl Database {
});
let mut schema = Schema::new();
let rows = conn.query("SELECT * FROM sqlite_schema")?;
if let Some(mut rows) = rows {
loop {
match rows.next_row()? {
RowResult::Row(row) => {
let ty = row.get::<&str>(0)?;
if ty != "table" && ty != "index" {
continue;
}
match ty {
"table" => {
let root_page: i64 = row.get::<i64>(3)?;
let sql: &str = row.get::<&str>(4)?;
let table = schema::BTreeTable::from_sql(sql, root_page as usize)?;
schema.add_table(Rc::new(table));
}
"index" => {
let root_page: i64 = row.get::<i64>(3)?;
match row.get::<&str>(4) {
Ok(sql) => {
let index =
schema::Index::from_sql(sql, root_page as usize)?;
schema.add_index(Rc::new(index));
}
_ => continue,
// TODO parse auto index structures
}
}
_ => continue,
}
}
RowResult::IO => {
// TODO: How do we ensure that the I/O we submitted to
// read the schema is actually complete?
io.run_once()?;
}
RowResult::Done => break,
}
}
}
let schema = Rc::new(schema);
parse_schema_rows(rows, &mut schema, io)?;
let schema = Rc::new(RefCell::new(schema));
let header = db_header;
Ok(Rc::new(Database {
pager,
@@ -208,7 +171,7 @@ pub fn maybe_init_database_file(file: &Rc<dyn File>, io: &Arc<dyn IO>) -> Result
pub struct Connection {
pager: Rc<Pager>,
schema: Rc<Schema>,
schema: Rc<RefCell<Schema>>,
header: Rc<RefCell<DatabaseHeader>>,
db: Weak<Database>, // backpointer to the database holding this connection
}
@@ -223,7 +186,7 @@ impl Connection {
match cmd {
Cmd::Stmt(stmt) => {
let program = Rc::new(translate::translate(
&self.schema,
&*self.schema.borrow(),
stmt,
self.header.clone(),
self.pager.clone(),
@@ -248,7 +211,7 @@ impl Connection {
match cmd {
Cmd::Stmt(stmt) => {
let program = Rc::new(translate::translate(
&self.schema,
&*self.schema.borrow(),
stmt,
self.header.clone(),
self.pager.clone(),
@@ -259,7 +222,7 @@ impl Connection {
}
Cmd::Explain(stmt) => {
let program = translate::translate(
&self.schema,
&*self.schema.borrow(),
stmt,
self.header.clone(),
self.pager.clone(),
@@ -271,7 +234,7 @@ impl Connection {
Cmd::ExplainQueryPlan(stmt) => {
match stmt {
ast::Stmt::Select(select) => {
let plan = prepare_select_plan(&self.schema, select)?;
let plan = prepare_select_plan(&*self.schema.borrow(), select)?;
let (plan, _) = optimize_plan(plan)?;
println!("{}", plan);
}
@@ -293,7 +256,7 @@ impl Connection {
match cmd {
Cmd::Explain(stmt) => {
let program = translate::translate(
&self.schema,
&*self.schema.borrow(),
stmt,
self.header.clone(),
self.pager.clone(),
@@ -304,7 +267,7 @@ impl Connection {
Cmd::ExplainQueryPlan(_stmt) => todo!(),
Cmd::Stmt(stmt) => {
let program = translate::translate(
&self.schema,
&*self.schema.borrow(),
stmt,
self.header.clone(),
self.pager.clone(),

4
core/pseudo.rs
View File

@@ -87,4 +87,8 @@ impl Cursor for PseudoCursor {
let _ = key;
todo!()
}
fn btree_create(&mut self, _flags: usize) -> u32 {
unreachable!("Please don't.")
}
}

25
core/storage/btree.rs
View File

@@ -648,7 +648,7 @@ impl BTreeCursor {
pointer_area_pc_by_idx + 2,
);
}
page.write_u16(pointer_area_pc_by_idx, pc);
page.write_u16(pointer_area_pc_by_idx - page.offset, pc);
// update first byte of content area
page.write_u16(BTREE_HEADER_OFFSET_CELL_CONTENT, pc);
@@ -1128,17 +1128,13 @@ impl BTreeCursor {
if gap + 2 + amount > top {
// defragment
self.defragment_page(page_ref, RefCell::borrow(&self.database_header));
let buf = page_ref.as_ptr();
top = u16::from_be_bytes([buf[5], buf[6]]) as usize;
top = page_ref.read_u16(BTREE_HEADER_OFFSET_CELL_CONTENT) as usize;
}
let db_header = RefCell::borrow(&self.database_header);
top -= amount;
{
let buf = page_ref.as_ptr();
buf[5..7].copy_from_slice(&(top as u16).to_be_bytes());
}
page_ref.write_u16(BTREE_HEADER_OFFSET_CELL_CONTENT, top as u16);
let usable_space = (db_header.page_size - db_header.unused_space as u16) as usize;
assert!(top + amount <= usable_space);
@@ -1358,6 +1354,7 @@ impl BTreeCursor {
let id = page.id as u32;
let contents = page.contents.as_mut().unwrap();
// TODO: take into account offset here?
let buf = contents.as_ptr();
let as_bytes = id.to_be_bytes();
// update pointer to new overflow page
@@ -1679,6 +1676,20 @@ impl Cursor for BTreeCursor {
Ok(CursorResult::Ok(equals))
}
}
fn btree_create(&mut self, flags: usize) -> u32 {
let page_type = match flags {
1 => PageType::TableLeaf,
2 => PageType::IndexLeaf,
_ => unreachable!(
"wrong create table falgs, should be 1 for table and 2 for index, got {}",
flags,
),
};
let page = self.allocate_page(page_type);
let id = page.borrow().id;
id as u32
}
}
pub fn btree_init_page(page: &Rc<RefCell<Page>>, page_type: PageType, db_header: &DatabaseHeader) {

227
core/translate/mod.rs
View File

@@ -16,6 +16,7 @@ pub(crate) mod planner;
pub(crate) mod select;
use std::cell::RefCell;
use std::fmt::Display;
use std::rc::{Rc, Weak};
use crate::schema::Schema;
@@ -26,6 +27,7 @@ use crate::{bail_parse_error, Connection, Result};
use insert::translate_insert;
use select::translate_select;
use sqlite3_parser::ast;
use sqlite3_parser::ast::fmt::ToTokens;
/// Translate SQL statement into bytecode program.
pub fn translate(
@@ -42,7 +44,24 @@ pub fn translate(
ast::Stmt::Begin(_, _) => bail_parse_error!("BEGIN not supported yet"),
ast::Stmt::Commit(_) => bail_parse_error!("COMMIT not supported yet"),
ast::Stmt::CreateIndex { .. } => bail_parse_error!("CREATE INDEX not supported yet"),
ast::Stmt::CreateTable { .. } => bail_parse_error!("CREATE TABLE not supported yet"),
ast::Stmt::CreateTable {
temporary,
if_not_exists,
tbl_name,
body,
} => {
if temporary {
bail_parse_error!("TEMPORARY table not supported yet");
}
translate_create_table(
tbl_name,
body,
if_not_exists,
database_header,
connection,
schema,
)
}
ast::Stmt::CreateTrigger { .. } => bail_parse_error!("CREATE TRIGGER not supported yet"),
ast::Stmt::CreateView { .. } => bail_parse_error!("CREATE VIEW not supported yet"),
ast::Stmt::CreateVirtualTable { .. } => {
@@ -85,6 +104,188 @@ pub fn translate(
}
}
/* Example:
sqlite> EXPLAIN CREATE TABLE users (id INT, email TEXT);;
addr opcode p1 p2 p3 p4 p5 comment
---- ------------- ---- ---- ---- ------------- -- -------------
0 Init 0 30 0 0 Start at 30
1 ReadCookie 0 3 2 0
2 If 3 5 0 0
3 SetCookie 0 2 4 0
4 SetCookie 0 5 1 0
5 CreateBtree 0 2 1 0 r[2]=root iDb=0 flags=1
6 OpenWrite 0 1 0 5 0 root=1 iDb=0
7 NewRowid 0 1 0 0 r[1]=rowid
8 Blob 6 3 0 0 r[3]= (len=6)
9 Insert 0 3 1 8 intkey=r[1] data=r[3]
10 Close 0 0 0 0
11 Close 0 0 0 0
12 Null 0 4 5 0 r[4..5]=NULL
13 Noop 2 0 4 0
14 OpenWrite 1 1 0 5 0 root=1 iDb=0; sqlite_master
15 SeekRowid 1 17 1 0 intkey=r[1]
16 Rowid 1 5 0 0 r[5]= rowid of 1
17 IsNull 5 26 0 0 if r[5]==NULL goto 26
18 String8 0 6 0 table 0 r[6]='table'
19 String8 0 7 0 users 0 r[7]='users'
20 String8 0 8 0 users 0 r[8]='users'
21 Copy 2 9 0 0 r[9]=r[2]
22 String8 0 10 0 CREATE TABLE users (id INT, email TEXT) 0 r[10]='CREATE TABLE users (id INT, email TEXT)'
23 MakeRecord 6 5 4 BBBDB 0 r[4]=mkrec(r[6..10])
24 Delete 1 68 5 0
25 Insert 1 4 5 0 intkey=r[5] data=r[4]
26 SetCookie 0 1 1 0
27 ParseSchema 0 0 0 tbl_name='users' AND type!='trigger' 0
28 SqlExec 1 0 0 PRAGMA "main".integrity_check('users') 0
29 Halt 0 0 0 0
30 Transaction 0 1 0 0 1 usesStmtJournal=1
31 Goto 0 1 0 0
*/
fn translate_create_table(
tbl_name: ast::QualifiedName,
body: ast::CreateTableBody,
if_not_exists: bool,
database_header: Rc<RefCell<DatabaseHeader>>,
connection: Weak<Connection>,
schema: &Schema,
) -> Result<Program> {
let mut program = ProgramBuilder::new();
if schema.get_table(tbl_name.name.0.as_str()).is_some() {
if if_not_exists {
let init_label = program.allocate_label();
program.emit_insn_with_label_dependency(
Insn::Init {
target_pc: init_label,
},
init_label,
);
let start_offset = program.offset();
program.emit_insn(Insn::Halt {
err_code: 0,
description: String::new(),
});
program.resolve_label(init_label, program.offset());
program.emit_insn(Insn::Transaction { write: true });
program.emit_constant_insns();
program.emit_insn(Insn::Goto {
target_pc: start_offset,
});
return Ok(program.build(database_header, connection));
}
bail_parse_error!("Table {} already exists", tbl_name);
}
let sql = create_table_body_to_str(&tbl_name, &body);
let parse_schema_label = program.allocate_label();
let init_label = program.allocate_label();
program.emit_insn_with_label_dependency(
Insn::Init {
target_pc: init_label,
},
init_label,
);
let start_offset = program.offset();
// TODO: ReadCookie
// TODO: If
// TODO: SetCookie
// TODO: SetCookie
let root_reg = program.alloc_register();
program.emit_insn(Insn::CreateBtree {
db: 0,
root: root_reg,
flags: 1,
});
let table_id = "sqlite_schema".to_string();
let table = schema.get_table(&table_id).unwrap();
let table = crate::schema::Table::BTree(table.clone());
let sqlite_schema_cursor_id =
program.alloc_cursor_id(Some(table_id.to_owned()), Some(table.to_owned()));
program.emit_insn(Insn::OpenWriteAsync {
cursor_id: sqlite_schema_cursor_id,
root_page: 1,
});
program.emit_insn(Insn::OpenWriteAwait {});
let rowid_reg = program.alloc_register();
program.emit_insn(Insn::NewRowid {
cursor: sqlite_schema_cursor_id,
rowid_reg,
prev_largest_reg: 0,
});
let null_reg_1 = program.alloc_register();
let null_reg_2 = program.alloc_register();
program.emit_insn(Insn::Null {
dest: null_reg_1,
dest_end: Some(null_reg_2),
});
let type_reg = program.alloc_register();
program.emit_insn(Insn::String8 {
value: "table".to_string(),
dest: type_reg,
});
let name_reg = program.alloc_register();
program.emit_insn(Insn::String8 {
value: tbl_name.name.0.to_string(),
dest: name_reg,
});
let tbl_name_reg = program.alloc_register();
program.emit_insn(Insn::String8 {
value: tbl_name.name.0.to_string(),
dest: tbl_name_reg,
});
let rootpage_reg = program.alloc_register();
program.emit_insn(Insn::Copy {
src_reg: root_reg,
dst_reg: rootpage_reg,
amount: 1,
});
let sql_reg = program.alloc_register();
program.emit_insn(Insn::String8 {
value: sql.to_string(),
dest: sql_reg,
});
let record_reg = program.alloc_register();
program.emit_insn(Insn::MakeRecord {
start_reg: type_reg,
count: 5,
dest_reg: record_reg,
});
// TODO: Delete
program.emit_insn(Insn::InsertAsync {
cursor: sqlite_schema_cursor_id,
key_reg: rowid_reg,
record_reg,
flag: 0,
});
program.emit_insn(Insn::InsertAwait {
cursor_id: sqlite_schema_cursor_id,
});
program.resolve_label(parse_schema_label, program.offset());
// TODO: SetCookie
//
// TODO: remove format, it sucks for performance but is convinient
let parse_schema_where_clause = format!("tbl_name = '{}' AND type != 'trigger'", tbl_name);
program.emit_insn(Insn::ParseSchema {
db: sqlite_schema_cursor_id,
where_clause: parse_schema_where_clause,
});
// TODO: SqlExec
program.emit_insn(Insn::Halt {
err_code: 0,
description: String::new(),
});
program.resolve_label(init_label, program.offset());
program.emit_insn(Insn::Transaction { write: true });
program.emit_constant_insns();
program.emit_insn(Insn::Goto {
target_pc: start_offset,
});
Ok(program.build(database_header, connection))
}
fn translate_pragma(
name: &ast::QualifiedName,
body: Option<ast::PragmaBody>,
@@ -187,3 +388,27 @@ fn update_pragma(name: &str, value: i64, header: Rc<RefCell<DatabaseHeader>>, pa
_ => todo!(),
}
}
struct TableFormatter<'a> {
body: &'a ast::CreateTableBody,
}
impl<'a> Display for TableFormatter<'a> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.body.to_fmt(f)
}
}
fn create_table_body_to_str(tbl_name: &ast::QualifiedName, body: &ast::CreateTableBody) -> String {
let mut sql = String::new();
let formatter = TableFormatter { body };
sql.push_str(format!("CREATE TABLE {} {}", tbl_name.name.0, formatter).as_str());
match body {
ast::CreateTableBody::ColumnsAndConstraints {
columns: _,
constraints: _,
options: _,
} => {}
ast::CreateTableBody::AsSelect(_select) => todo!("as select not yet supported"),
}
sql
}

2
core/types.rs
View File

@@ -373,6 +373,7 @@ impl OwnedRecord {
let mut header_size = buf.len() - initial_i;
// write content
for value in &self.values {
// TODO: make integers and floats with smaller serial types
match value {
OwnedValue::Null => {}
OwnedValue::Integer(i) => buf.extend_from_slice(&i.to_be_bytes()),
@@ -440,4 +441,5 @@ pub trait Cursor {
fn exists(&mut self, key: &OwnedValue) -> Result<CursorResult<bool>>;
fn set_null_flag(&mut self, flag: bool);
fn get_null_flag(&self) -> bool;
fn btree_create(&mut self, flags: usize) -> u32;
}

49
core/util.rs
View File

@@ -1,3 +1,10 @@
use std::{rc::Rc, sync::Arc};
use crate::{
schema::{self, Schema},
Result, RowResult, Rows, IO,
};
pub fn normalize_ident(ident: &str) -> String {
(if ident.starts_with('"') && ident.ends_with('"') {
&ident[1..ident.len() - 1]
@@ -6,3 +13,45 @@ pub fn normalize_ident(ident: &str) -> String {
})
.to_lowercase()
}
pub fn parse_schema_rows(rows: Option<Rows>, schema: &mut Schema, io: Arc<dyn IO>) -> Result<()> {
if let Some(mut rows) = rows {
loop {
match rows.next_row()? {
RowResult::Row(row) => {
let ty = row.get::<&str>(0)?;
if ty != "table" && ty != "index" {
continue;
}
match ty {
"table" => {
let root_page: i64 = row.get::<i64>(3)?;
let sql: &str = row.get::<&str>(4)?;
let table = schema::BTreeTable::from_sql(sql, root_page as usize)?;
schema.add_table(Rc::new(table));
}
"index" => {
let root_page: i64 = row.get::<i64>(3)?;
match row.get::<&str>(4) {
Ok(sql) => {
let index = schema::Index::from_sql(sql, root_page as usize)?;
schema.add_index(Rc::new(index));
}
_ => continue,
// TODO parse auto index structures
}
}
_ => continue,
}
}
RowResult::IO => {
// TODO: How do we ensure that the I/O we submitted to
// read the schema is actually complete?
io.run_once()?;
}
RowResult::Done => break,
}
}
}
Ok(())
}

4
core/vdbe/builder.rs
View File

@@ -319,6 +319,10 @@ impl ProgramBuilder {
assert!(*target_pc < 0);
*target_pc = to_offset;
}
Insn::IsNull { src: _, target_pc } => {
assert!(*target_pc < 0);
*target_pc = to_offset;
}
_ => {
todo!("missing resolve_label for {:?}", insn);
}

36
core/vdbe/explain.rs
View File

@@ -875,6 +875,42 @@ pub fn insn_to_str(
0,
format!("r[{}]=r[{}]", dst_reg, src_reg),
),
Insn::CreateBtree { db, root, flags } => (
"CreateBtree",
*db as i32,
*root as i32,
*flags as i32,
OwnedValue::Text(Rc::new("".to_string())),
0,
format!("r[{}]=root iDb={} flags={}", root, db, flags),
),
Insn::Close { cursor_id } => (
"Close",
*cursor_id as i32,
0,
0,
OwnedValue::Text(Rc::new("".to_string())),
0,
"".to_string(),
),
Insn::IsNull { src, target_pc } => (
"IsNull",
*src as i32,
*target_pc as i32,
0,
OwnedValue::Text(Rc::new("".to_string())),
0,
format!("if (r[{}]==NULL) goto {}", src, target_pc),
),
Insn::ParseSchema { db, where_clause } => (
"ParseSchema",
*db as i32,
0,
0,
OwnedValue::Text(Rc::new(where_clause.clone())),
0,
where_clause.clone(),
),
};
format!(
"{:<4} {:<17} {:<4} {:<4} {:<4} {:<13} {:<2} {}",

77
core/vdbe/mod.rs
View File

@@ -32,10 +32,11 @@ use crate::storage::{btree::BTreeCursor, pager::Pager};
use crate::types::{
AggContext, Cursor, CursorResult, OwnedRecord, OwnedValue, Record, SeekKey, SeekOp,
};
use crate::DATABASE_VERSION;
use crate::util::parse_schema_rows;
#[cfg(feature = "json")]
use crate::{function::JsonFunc, json::get_json};
use crate::{Connection, Result, TransactionState};
use crate::{Rows, DATABASE_VERSION};
use datetime::{exec_date, exec_time, exec_unixepoch};
@@ -475,6 +476,34 @@ pub enum Insn {
dst_reg: usize,
amount: usize, // 0 amount means we include src_reg, dst_reg..=dst_reg+amount = src_reg..=src_reg+amount
},
/// Allocate a new b-tree.
CreateBtree {
/// Allocate b-tree in main database if zero or in temp database if non-zero (P1).
db: usize,
/// The root page of the new b-tree (P2).
root: usize,
/// Flags (P3).
flags: usize,
},
/// Close a cursor.
Close {
cursor_id: CursorID,
},
/// Check if the register is null.
IsNull {
/// Source register (P1).
src: usize,
/// Jump to this PC if the register is null (P2).
target_pc: BranchOffset,
},
ParseSchema {
db: usize,
where_clause: String,
},
}
// Index of insn in list of insns
@@ -2151,6 +2180,48 @@ impl Program {
}
state.pc += 1;
}
Insn::CreateBtree { db, root, flags: _ } => {
if *db > 0 {
// TODO: implement temp datbases
todo!("temp databases not implemented yet");
}
let mut cursor = Box::new(BTreeCursor::new(
pager.clone(),
0,
self.database_header.clone(),
));
let root_page = cursor.btree_create(1);
state.registers[*root] = OwnedValue::Integer(root_page as i64);
state.pc += 1;
}
Insn::Close { cursor_id } => {
cursors.remove(cursor_id);
state.pc += 1;
}
Insn::IsNull { src, target_pc } => {
if matches!(state.registers[*src], OwnedValue::Null) {
state.pc = *target_pc;
} else {
state.pc += 1;
}
}
Insn::ParseSchema {
db: _,
where_clause,
} => {
let conn = self.connection.upgrade();
let conn = conn.as_ref().unwrap();
let stmt = conn.prepare(format!(
"SELECT * FROM sqlite_schema WHERE {}",
where_clause
))?;
let rows = Rows { stmt };
let mut schema = RefCell::borrow_mut(&conn.schema);
// TODO: This function below is synchronous, make it not async
parse_schema_rows(Some(rows), &mut *schema, conn.pager.io.clone())?;
state.pc += 1;
}
}
}
}
@@ -2817,6 +2888,10 @@ mod tests {
fn exists(&mut self, _key: &OwnedValue) -> Result<CursorResult<bool>> {
unimplemented!()
}
fn btree_create(&mut self, _flags: usize) -> u32 {
unimplemented!()
}
}
#[test]

4
core/vdbe/sorter.rs
View File

@@ -87,4 +87,8 @@ impl Cursor for Sorter {
let _ = key;
todo!()
}
fn btree_create(&mut self, _flags: usize) -> u32 {
unreachable!("Why did you try to build a new tree with a sorter??? Stand up, open the door and take a walk for 30 min to come back with a better plan.");
}
}