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

Merge branch 'main' of https://github.com/tursodatabase/limbo

Browse Source
This commit is contained in:
alpaylan
2025-02-08 09:35:21 -05:00
parent 6308ce4544 a1a9218131
commit 53ce08aefa
9 changed files with 211 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
Cargo.lock generated
View File

@@ -1394,15 +1394,6 @@ version = "1.0.14"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d75a2a4b1b190afb6f5425f10f6a8f959d2ea0b9c2b1d79553551850539e4674"
[[package]]
name = "java-limbo"
version = "0.0.14"
dependencies = [
"jni",
"limbo_core",
"thiserror 2.0.11",
]
[[package]]
name = "jni"
version = "0.21.1"
@@ -1590,6 +1581,15 @@ dependencies = [
"limbo_core",
]
[[package]]
name = "limbo-java"
version = "0.0.14"
dependencies = [
"jni",
"limbo_core",
"thiserror 2.0.11",
]
[[package]]
name = "limbo-wasm"
version = "0.0.14"

5
README.md
View File

@@ -142,6 +142,11 @@ for rows.Next() {
}
```
### ☕️ Java (wip)
We integrated Limbo into JDBC. For detailed instructions on how to use Limbo with java, please refer to
the [README.md under bindings/java](bindings/java/README.md).
## Contributing
We'd love to have you contribute to Limbo! Please check out the [contribution guide] to get started.

2
bindings/java/Cargo.toml
View File

@@ -1,5 +1,5 @@
[package]
name = "java-limbo"
name = "limbo-java"
version.workspace = true
authors.workspace = true
edition.workspace = true

53
cli/app.rs
View File

@@ -2,7 +2,8 @@ use crate::{
import::{ImportFile, IMPORT_HELP},
opcodes_dictionary::OPCODE_DESCRIPTIONS,
};
use cli_table::{Cell, Table};
use cli_table::format::{Border, HorizontalLine, Separator, VerticalLine};
use cli_table::{Cell, Style, Table};
use limbo_core::{Database, LimboError, Statement, StepResult, Value};
use clap::{Parser, ValueEnum};
@@ -670,6 +671,16 @@ impl Limbo {
return Ok(());
}
let mut table_rows: Vec<Vec<_>> = vec![];
if rows.num_columns() > 0 {
let columns = (0..rows.num_columns())
.map(|i| {
rows.get_column_name(i)
.map(|name| name.cell().bold(true))
.unwrap_or_else(|| " ".cell())
})
.collect::<Vec<_>>();
table_rows.push(columns);
}
loop {
match rows.step() {
Ok(StepResult::Row) => {
@@ -707,11 +718,7 @@ impl Limbo {
}
}
}
if let Ok(table) = table_rows.table().display() {
let _ = self.write_fmt(format_args!("{}", table));
} else {
let _ = self.writeln("Error displaying table.");
}
self.print_table(table_rows);
}
},
Ok(None) => {}
@@ -727,6 +734,40 @@ impl Limbo {
Ok(())
}
fn print_table(&mut self, table_rows: Vec<Vec<cli_table::CellStruct>>) {
if table_rows.is_empty() {
return;
}
let horizontal_line = HorizontalLine::new('┌', '┐', '┬', '─');
let horizontal_line_mid = HorizontalLine::new('├', '┤', '┼', '─');
let horizontal_line_bottom = HorizontalLine::new('└', '┘', '┴', '─');
let vertical_line = VerticalLine::new('│');
let border = Border::builder()
.top(horizontal_line)
.bottom(horizontal_line_bottom)
.left(vertical_line.clone())
.right(vertical_line.clone())
.build();
let separator = Separator::builder()
.column(Some(vertical_line))
.row(Some(horizontal_line_mid))
.build();
if let Ok(table) = table_rows
.table()
.border(border)
.separator(separator)
.display()
{
let _ = self.write_fmt(format_args!("{}", table));
} else {
let _ = self.writeln("Error displaying table.");
}
}
fn display_schema(&mut self, table: Option<&str>) -> anyhow::Result<()> {
let sql = match table {
Some(table_name) => format!(

19
core/io/io_uring.rs
View File

@@ -4,7 +4,6 @@ use log::{debug, trace};
use rustix::fs::{self, FlockOperation, OFlags};
use rustix::io_uring::iovec;
use std::cell::RefCell;
use std::collections::HashMap;
use std::fmt;
use std::io::ErrorKind;
use std::os::fd::AsFd;
@@ -40,7 +39,7 @@ pub struct UringIO {
struct WrappedIOUring {
ring: io_uring::IoUring,
pending_ops: usize,
pub pending: HashMap<u64, Rc<Completion>>,
pub pending: [Option<Rc<Completion>>; MAX_IOVECS as usize + 1],
key: u64,
}
@@ -63,7 +62,7 @@ impl UringIO {
ring: WrappedIOUring {
ring,
pending_ops: 0,
pending: HashMap::new(),
pending: [const { None }; MAX_IOVECS as usize + 1],
key: 0,
},
iovecs: [iovec {
@@ -92,7 +91,7 @@ impl InnerUringIO {
impl WrappedIOUring {
fn submit_entry(&mut self, entry: &io_uring::squeue::Entry, c: Rc<Completion>) {
trace!("submit_entry({:?})", entry);
self.pending.insert(entry.get_user_data(), c);
self.pending[entry.get_user_data() as usize] = Some(c);
unsafe {
self.ring
.submission()
@@ -124,6 +123,11 @@ impl WrappedIOUring {
fn get_key(&mut self) -> u64 {
self.key += 1;
if self.key == MAX_IOVECS as u64 {
let key = self.key;
self.key = 0;
return key;
}
self.key
}
}
@@ -175,10 +179,11 @@ impl IO for UringIO {
)));
}
{
let c = ring.pending.get(&cqe.user_data()).unwrap().clone();
c.complete(cqe.result());
if let Some(c) = ring.pending[cqe.user_data() as usize].as_ref() {
c.complete(cqe.result());
}
}
ring.pending.remove(&cqe.user_data());
ring.pending[cqe.user_data() as usize] = None;
}
Ok(())
}

162
core/mvcc/database/mod.rs
View File

@@ -58,7 +58,9 @@ impl<T> LogRecord<T> {
/// versions switch to tracking timestamps.
#[derive(Clone, Debug, PartialEq, PartialOrd)]
enum TxTimestampOrID {
/// A committed transaction's timestamp.
Timestamp(u64),
/// The ID of a non-committed transaction.
TxID(TxID),
}
@@ -229,55 +231,6 @@ impl<Clock: LogicalClock, T: Sync + Send + Clone + Debug + 'static> MvStore<Cloc
}
}
// Extracts the begin timestamp from a transaction
fn get_begin_timestamp(&self, ts_or_id: &TxTimestampOrID) -> u64 {
match ts_or_id {
TxTimestampOrID::Timestamp(ts) => *ts,
TxTimestampOrID::TxID(tx_id) => {
self.txs
.get(tx_id)
.unwrap()
.value()
.read()
.unwrap()
.begin_ts
}
}
}
/// Inserts a new row version into the database, while making sure that
/// the row version is inserted in the correct order.
fn insert_version(&self, id: RowID, row_version: RowVersion<T>) {
let versions = self.rows.get_or_insert_with(id, || RwLock::new(Vec::new()));
let mut versions = versions.value().write().unwrap();
self.insert_version_raw(&mut versions, row_version)
}
/// Inserts a new row version into the internal data structure for versions,
/// while making sure that the row version is inserted in the correct order.
fn insert_version_raw(&self, versions: &mut Vec<RowVersion<T>>, row_version: RowVersion<T>) {
// NOTICE: this is an insert a'la insertion sort, with pessimistic linear complexity.
// However, we expect the number of versions to be nearly sorted, so we deem it worthy
// to search linearly for the insertion point instead of paying the price of using
// another data structure, e.g. a BTreeSet. If it proves to be too quadratic empirically,
// we can either switch to a tree-like structure, or at least use partition_point()
// which performs a binary search for the insertion point.
let position = versions
.iter()
.rposition(|v| {
self.get_begin_timestamp(&v.begin) < self.get_begin_timestamp(&row_version.begin)
})
.map(|p| p + 1)
.unwrap_or(0);
if versions.len() - position > 3 {
tracing::debug!(
"Inserting a row version {} positions from the end",
versions.len() - position
);
}
versions.insert(position, row_version);
}
/// Inserts a new row into the database.
///
/// This function inserts a new `row` into the database within the context
@@ -365,6 +318,10 @@ impl<Clock: LogicalClock, T: Sync + Send + Clone + Debug + 'static> MvStore<Cloc
.ok_or(DatabaseError::NoSuchTransactionID(tx_id))?;
let tx = tx.value().read().unwrap();
assert_eq!(tx.state, TransactionState::Active);
let version_is_visible_to_current_tx = is_version_visible(&self.txs, &tx, rv);
if !version_is_visible_to_current_tx {
continue;
}
if is_write_write_conflict(&self.txs, &tx, rv) {
drop(row_versions);
drop(row_versions_opt);
@@ -372,19 +329,18 @@ impl<Clock: LogicalClock, T: Sync + Send + Clone + Debug + 'static> MvStore<Cloc
self.rollback_tx(tx_id);
return Err(DatabaseError::WriteWriteConflict);
}
if is_version_visible(&self.txs, &tx, rv) {
rv.end = Some(TxTimestampOrID::TxID(tx.tx_id));
drop(row_versions);
drop(row_versions_opt);
drop(tx);
let tx = self
.txs
.get(&tx_id)
.ok_or(DatabaseError::NoSuchTransactionID(tx_id))?;
let mut tx = tx.value().write().unwrap();
tx.insert_to_write_set(id);
return Ok(true);
}
rv.end = Some(TxTimestampOrID::TxID(tx.tx_id));
drop(row_versions);
drop(row_versions_opt);
drop(tx);
let tx = self
.txs
.get(&tx_id)
.ok_or(DatabaseError::NoSuchTransactionID(tx_id))?;
let mut tx = tx.value().write().unwrap();
tx.insert_to_write_set(id);
return Ok(true);
}
}
Ok(false)
@@ -556,7 +512,6 @@ impl<Clock: LogicalClock, T: Sync + Send + Clone + Debug + 'static> MvStore<Cloc
*/
tx.state.store(TransactionState::Committed(end_ts));
tracing::trace!("COMMIT {tx}");
let tx_begin_ts = tx.begin_ts;
let write_set: Vec<RowID> = tx.write_set.iter().map(|v| *v.value()).collect();
drop(tx);
// Postprocessing: inserting row versions and logging the transaction to persistent storage.
@@ -568,7 +523,9 @@ impl<Clock: LogicalClock, T: Sync + Send + Clone + Debug + 'static> MvStore<Cloc
for row_version in row_versions.iter_mut() {
if let TxTimestampOrID::TxID(id) = row_version.begin {
if id == tx_id {
row_version.begin = TxTimestampOrID::Timestamp(tx_begin_ts);
// New version is valid STARTING FROM committing transaction's end timestamp
// See diagram on page 299: https://www.cs.cmu.edu/~15721-f24/papers/Hekaton.pdf
row_version.begin = TxTimestampOrID::Timestamp(end_ts);
self.insert_version_raw(
&mut log_record.row_versions,
row_version.clone(),
@@ -577,6 +534,8 @@ impl<Clock: LogicalClock, T: Sync + Send + Clone + Debug + 'static> MvStore<Cloc
}
if let Some(TxTimestampOrID::TxID(id)) = row_version.end {
if id == tx_id {
// New version is valid UNTIL committing transaction's end timestamp
// See diagram on page 299: https://www.cs.cmu.edu/~15721-f24/papers/Hekaton.pdf
row_version.end = Some(TxTimestampOrID::Timestamp(end_ts));
self.insert_version_raw(
&mut log_record.row_versions,
@@ -718,10 +677,69 @@ impl<Clock: LogicalClock, T: Sync + Send + Clone + Debug + 'static> MvStore<Cloc
}
Ok(())
}
// Extracts the begin timestamp from a transaction
fn get_begin_timestamp(&self, ts_or_id: &TxTimestampOrID) -> u64 {
match ts_or_id {
TxTimestampOrID::Timestamp(ts) => *ts,
TxTimestampOrID::TxID(tx_id) => {
self.txs
.get(tx_id)
.unwrap()
.value()
.read()
.unwrap()
.begin_ts
}
}
}
/// Inserts a new row version into the database, while making sure that
/// the row version is inserted in the correct order.
fn insert_version(&self, id: RowID, row_version: RowVersion<T>) {
let versions = self.rows.get_or_insert_with(id, || RwLock::new(Vec::new()));
let mut versions = versions.value().write().unwrap();
self.insert_version_raw(&mut versions, row_version)
}
/// Inserts a new row version into the internal data structure for versions,
/// while making sure that the row version is inserted in the correct order.
fn insert_version_raw(&self, versions: &mut Vec<RowVersion<T>>, row_version: RowVersion<T>) {
// NOTICE: this is an insert a'la insertion sort, with pessimistic linear complexity.
// However, we expect the number of versions to be nearly sorted, so we deem it worthy
// to search linearly for the insertion point instead of paying the price of using
// another data structure, e.g. a BTreeSet. If it proves to be too quadratic empirically,
// we can either switch to a tree-like structure, or at least use partition_point()
// which performs a binary search for the insertion point.
let position = versions
.iter()
.rposition(|v| {
self.get_begin_timestamp(&v.begin) < self.get_begin_timestamp(&row_version.begin)
})
.map(|p| p + 1)
.unwrap_or(0);
if versions.len() - position > 3 {
tracing::debug!(
"Inserting a row version {} positions from the end",
versions.len() - position
);
}
versions.insert(position, row_version);
}
}
/// A write-write conflict happens when transaction T_m attempts to update a
/// row version that is currently being updated by an active transaction T_n.
/// A write-write conflict happens when transaction T_current attempts to update a
/// row version that is:
/// a) currently being updated by an active transaction T_previous, or
/// b) was updated by an ended transaction T_previous that committed AFTER T_current started
/// but BEFORE T_previous commits.
///
/// "Suppose transaction T wants to update a version V. V is updatable
/// only if it is the latest version, that is, it has an end timestamp equal
/// to infinity or its End field contains the ID of a transaction TE and
/// TE’s state is Aborted"
/// Ref: https://www.cs.cmu.edu/~15721-f24/papers/Hekaton.pdf , page 301,
/// 2.6. Updating a Version.
pub(crate) fn is_write_write_conflict<T>(
txs: &SkipMap<TxID, RwLock<Transaction>>,
tx: &Transaction,
@@ -731,12 +749,16 @@ pub(crate) fn is_write_write_conflict<T>(
Some(TxTimestampOrID::TxID(rv_end)) => {
let te = txs.get(&rv_end).unwrap();
let te = te.value().read().unwrap();
match te.state.load() {
TransactionState::Active | TransactionState::Preparing => tx.tx_id != te.tx_id,
_ => false,
if te.tx_id == tx.tx_id {
return false;
}
te.state.load() != TransactionState::Aborted
}
Some(TxTimestampOrID::Timestamp(_)) => false,
// A non-"infinity" end timestamp (here modeled by Some(ts)) functions as a write lock
// on the row, so it can never be updated by another transaction.
// Ref: https://www.cs.cmu.edu/~15721-f24/papers/Hekaton.pdf , page 301,
// 2.6. Updating a Version.
Some(TxTimestampOrID::Timestamp(_)) => true,
None => false,
}
}

16
core/mvcc/database/tests.rs
View File

@@ -382,7 +382,7 @@ fn test_fuzzy_read() {
table_id: 1,
row_id: 1,
},
data: "Hello".to_string(),
data: "First".to_string(),
};
db.insert(tx1, tx1_row.clone()).unwrap();
let row = db
@@ -419,7 +419,7 @@ fn test_fuzzy_read() {
table_id: 1,
row_id: 1,
},
data: "World".to_string(),
data: "Second".to_string(),
};
db.update(tx3, tx3_row).unwrap();
db.commit_tx(tx3).unwrap();
@@ -436,6 +436,18 @@ fn test_fuzzy_read() {
.unwrap()
.unwrap();
assert_eq!(tx1_row, row);
// T2 tries to update the row, but fails because T3 has already committed an update to the row,
// so T2 trying to write would violate snapshot isolation if it succeeded.
let tx2_newrow = Row {
id: RowID {
table_id: 1,
row_id: 1,
},
data: "Third".to_string(),
};
let update_result = db.update(tx2, tx2_newrow);
assert_eq!(Err(DatabaseError::WriteWriteConflict), update_result);
}
#[test]

16
vendored/sqlite3-parser/src/lexer/sql/mod.rs
View File

@@ -30,6 +30,7 @@ pub struct Parser<'input> {
scanner: Scanner<Tokenizer>,
/// lemon parser
parser: yyParser<'input>,
had_error: bool,
}
impl<'input> Parser<'input> {
@@ -43,12 +44,14 @@ impl<'input> Parser<'input> {
input,
scanner,
parser,
had_error: false,
}
}
/// Parse new `input`
pub fn reset(&mut self, input: &'input [u8]) {
self.input = input;
self.scanner.reset();
self.had_error = false;
}
/// Current line position in input
pub fn line(&self) -> u64 {
@@ -182,6 +185,10 @@ impl FallibleIterator for Parser<'_> {
fn next(&mut self) -> Result<Option<Cmd>, Error> {
//print!("line: {}, column: {}: ", self.scanner.line(), self.scanner.column());
// if we have already encountered an error, return None to signal that to fallible_iterator that we are done parsing
if self.had_error {
return Ok(None);
}
self.parser.ctx.reset();
let mut last_token_parsed = TK_EOF;
let mut eof = false;
@@ -197,6 +204,7 @@ impl FallibleIterator for Parser<'_> {
if token_type == TK_ILLEGAL {
// break out of parsing loop and return error
self.parser.sqlite3ParserFinalize();
self.had_error = true;
return Err(Error::UnrecognizedToken(
Some((self.scanner.line(), self.scanner.column())),
Some(start.into()),
@@ -242,12 +250,18 @@ impl FallibleIterator for Parser<'_> {
self.parser
.sqlite3Parser(TK_SEMI, sentinel(self.input.len()))
);
if self.parser.ctx.error().is_some() {
self.had_error = true;
}
}
try_with_position!(
self.scanner,
self.parser
.sqlite3Parser(TK_EOF, sentinel(self.input.len()))
);
if self.parser.ctx.error().is_some() {
self.had_error = true;
}
}
self.parser.sqlite3ParserFinalize();
if let Some(e) = self.parser.ctx.error() {
@@ -256,6 +270,7 @@ impl FallibleIterator for Parser<'_> {
Some((self.scanner.line(), self.scanner.column())),
Some((self.offset() - 1).into()),
);
self.had_error = true;
return Err(err);
}
let cmd = self.parser.ctx.cmd();
@@ -266,6 +281,7 @@ impl FallibleIterator for Parser<'_> {
Some((self.scanner.line(), self.scanner.column())),
Some((self.offset() - 1).into()),
);
self.had_error = true;
return Err(err);
}
}

15
vendored/sqlite3-parser/src/lexer/sql/test.rs
View File

@@ -338,6 +338,21 @@ fn qualified_table_name_within_triggers() {
);
}
#[test]
fn select_from_error_stops_at_first_error() {
let mut parser = Parser::new(b"SELECT FROM foo;");
// First next() call should return the first syntax error
let err = parser.next().unwrap_err();
assert!(matches!(err, Error::ParserError(_, _, _)));
// Second next() call should return Ok(None) since parsing should have stopped
assert_eq!(parser.next().unwrap(), None);
// Third next() call should also return Ok(None)
assert_eq!(parser.next().unwrap(), None);
}
#[test]
fn indexed_by_clause_within_triggers() {
expect_parser_err_msg(