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

core: Convert Rc<Pager> to Arc<Pager>

Browse Source
This commit is contained in:
Pekka Enberg
2025-09-16 21:03:18 +03:00
parent 104b8dd083
commit 17e9f05ea4
14 changed files with 202 additions and 210 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
core/incremental/compiler.rs
View File

@@ -21,7 +21,6 @@ use crate::Pager;
use crate::{return_and_restore_if_io, return_if_io, LimboError, Result};
use std::collections::HashMap;
use std::fmt::{self, Display, Formatter};
use std::rc::Rc;
use std::sync::Arc;
// The state table has 5 columns: operator_id, zset_id, element_id, value, weight
@@ -435,7 +434,7 @@ impl DbspCircuit {
pub fn run_circuit(
&mut self,
execute_state: &mut ExecuteState,
pager: &Rc<Pager>,
pager: &Arc<Pager>,
state_cursors: &mut DbspStateCursors,
commit_operators: bool,
) -> Result<IOResult<Delta>> {
@@ -462,7 +461,7 @@ impl DbspCircuit {
/// * `execute_state` - State machine containing input deltas and tracking execution progress
pub fn execute(
&mut self,
pager: Rc<Pager>,
pager: Arc<Pager>,
execute_state: &mut ExecuteState,
) -> Result<IOResult<Delta>> {
if let Some(root_id) = self.root {
@@ -499,7 +498,7 @@ impl DbspCircuit {
pub fn commit(
&mut self,
input_data: HashMap<String, Delta>,
pager: Rc<Pager>,
pager: Arc<Pager>,
) -> Result<IOResult<Delta>> {
// No root means nothing to commit
if self.root.is_none() {
@@ -643,7 +642,7 @@ impl DbspCircuit {
fn execute_node(
&mut self,
node_id: usize,
pager: Rc<Pager>,
pager: Arc<Pager>,
execute_state: &mut ExecuteState,
commit_operators: bool,
cursors: &mut DbspStateCursors,
@@ -1320,7 +1319,6 @@ mod tests {
use crate::translate::logical::LogicalSchema;
use crate::util::IOExt;
use crate::{Database, MemoryIO, Pager, IO};
use std::rc::Rc;
use std::sync::Arc;
use turso_parser::ast;
use turso_parser::parser::Parser;
@@ -1416,7 +1414,7 @@ mod tests {
}};
}
fn setup_btree_for_circuit() -> (Rc<Pager>, usize, usize, usize) {
fn setup_btree_for_circuit() -> (Arc<Pager>, usize, usize, usize) {
let io: Arc<dyn IO> = Arc::new(MemoryIO::new());
let db = Database::open_file(io.clone(), ":memory:", false, false).unwrap();
let conn = db.connect().unwrap();
@@ -1579,7 +1577,7 @@ mod tests {
fn test_execute(
circuit: &mut DbspCircuit,
inputs: HashMap<String, Delta>,
pager: Rc<Pager>,
pager: Arc<Pager>,
) -> Result<Delta> {
let mut execute_state = ExecuteState::Init {
input_data: DeltaSet::from_map(inputs),
@@ -1593,7 +1591,7 @@ mod tests {
// Helper to get the committed BTree state from main_data_root
// This reads the actual persisted data from the BTree
#[cfg(test)]
fn get_current_state(pager: Rc<Pager>, circuit: &DbspCircuit) -> Result<Delta> {
fn get_current_state(pager: Arc<Pager>, circuit: &DbspCircuit) -> Result<Delta> {
let mut delta = Delta::new();
let main_data_root = circuit.main_data_root;

7
core/incremental/cursor.rs
View File

@@ -38,7 +38,7 @@ pub struct MaterializedViewCursor {
// Core components
btree_cursor: Box<BTreeCursor>,
view: Arc<Mutex<IncrementalView>>,
pager: Rc<Pager>,
pager: Arc<Pager>,
// Current changes that are uncommitted
uncommitted: RowKeyZSet,
@@ -65,7 +65,7 @@ impl MaterializedViewCursor {
pub fn new(
btree_cursor: Box<BTreeCursor>,
view: Arc<Mutex<IncrementalView>>,
pager: Rc<Pager>,
pager: Arc<Pager>,
tx_state: Rc<ViewTransactionState>,
) -> Result<Self> {
Ok(Self {
@@ -302,7 +302,6 @@ mod tests {
use super::*;
use crate::util::IOExt;
use crate::{Connection, Database, OpenFlags};
use std::rc::Rc;
use std::sync::Arc;
/// Helper to create a test connection with a table and materialized view
@@ -335,7 +334,7 @@ mod tests {
/// Helper to create a test cursor for the materialized view
fn create_test_cursor(
conn: &Arc<Connection>,
) -> Result<(MaterializedViewCursor, Rc<ViewTransactionState>, Rc<Pager>)> {
) -> Result<(MaterializedViewCursor, Rc<ViewTransactionState>, Arc<Pager>)> {
// Get the schema and view
let view_mutex = conn
.schema

3
core/incremental/expr_compiler.rs
View File

@@ -12,7 +12,6 @@ use crate::vdbe::insn::Insn;
use crate::vdbe::{Program, ProgramState, Register};
use crate::SymbolTable;
use crate::{CaptureDataChangesMode, Connection, QueryMode, Result, Value};
use std::rc::Rc;
use std::sync::Arc;
use turso_parser::ast::{Expr, Literal, Operator};
@@ -386,7 +385,7 @@ impl CompiledExpression {
}
/// Execute the compiled expression with the given input values
pub fn execute(&self, values: &[Value], pager: Rc<Pager>) -> Result<Value> {
pub fn execute(&self, values: &[Value], pager: Arc<Pager>) -> Result<Value> {
match &self.executor {
ExpressionExecutor::Trivial(trivial) => {
// Fast path: evaluate trivial expression inline

4
core/incremental/operator.rs
View File

@@ -2239,7 +2239,7 @@ mod tests {
use std::sync::{Arc, Mutex};
/// Create a test pager for operator tests with both table and index
fn create_test_pager() -> (std::rc::Rc<crate::Pager>, usize, usize) {
fn create_test_pager() -> (std::sync::Arc<crate::Pager>, usize, usize) {
let io: Arc<dyn IO> = Arc::new(MemoryIO::new());
let db = Database::open_file(io.clone(), ":memory:", false, false).unwrap();
let conn = db.connect().unwrap();
@@ -2270,7 +2270,7 @@ mod tests {
/// Returns a Delta with all the current aggregate values
fn get_current_state_from_btree(
agg: &AggregateOperator,
pager: &std::rc::Rc<crate::Pager>,
pager: &std::sync::Arc<crate::Pager>,
cursors: &mut DbspStateCursors,
) -> Delta {
let mut result = Delta::new();

6
core/incremental/view.rs
View File

@@ -384,7 +384,7 @@ impl IncrementalView {
pub fn execute_with_uncommitted(
&mut self,
uncommitted: DeltaSet,
pager: Rc<Pager>,
pager: Arc<Pager>,
execute_state: &mut crate::incremental::compiler::ExecuteState,
) -> crate::Result<crate::types::IOResult<Delta>> {
// Initialize execute_state with the input data
@@ -560,7 +560,7 @@ impl IncrementalView {
pub fn populate_from_table(
&mut self,
conn: &std::sync::Arc<crate::Connection>,
pager: &std::rc::Rc<crate::Pager>,
pager: &std::sync::Arc<crate::Pager>,
_btree_cursor: &mut BTreeCursor,
) -> crate::Result<IOResult<()>> {
// If already populated, return immediately
@@ -835,7 +835,7 @@ impl IncrementalView {
pub fn merge_delta(
&mut self,
delta_set: DeltaSet,
pager: std::rc::Rc<crate::Pager>,
pager: std::sync::Arc<crate::Pager>,
) -> crate::Result<IOResult<()>> {
// Early return if all deltas are empty
if delta_set.is_empty() {

22
core/lib.rs
View File

@@ -489,7 +489,7 @@ impl Database {
.get();
let conn = Arc::new(Connection {
_db: self.clone(),
pager: RefCell::new(Rc::new(pager)),
pager: RefCell::new(Arc::new(pager)),
schema: RefCell::new(
self.schema
.lock()
@@ -867,7 +867,7 @@ impl CaptureDataChangesMode {
struct DatabaseCatalog {
name_to_index: HashMap<String, usize>,
allocated: Vec<u64>,
index_to_data: HashMap<usize, (Arc<Database>, Rc<Pager>)>,
index_to_data: HashMap<usize, (Arc<Database>, Arc<Pager>)>,
}
#[allow(unused)]
@@ -896,7 +896,7 @@ impl DatabaseCatalog {
}
}
fn get_pager_by_index(&self, idx: &usize) -> Rc<Pager> {
fn get_pager_by_index(&self, idx: &usize) -> Arc<Pager> {
let (_db, pager) = self
.index_to_data
.get(idx)
@@ -912,7 +912,7 @@ impl DatabaseCatalog {
index
}
fn insert(&mut self, s: &str, data: (Arc<Database>, Rc<Pager>)) -> usize {
fn insert(&mut self, s: &str, data: (Arc<Database>, Arc<Pager>)) -> usize {
let idx = self.add(s);
self.index_to_data.insert(idx, data);
idx
@@ -972,7 +972,7 @@ impl DatabaseCatalog {
pub struct Connection {
_db: Arc<Database>,
pager: RefCell<Rc<Pager>>,
pager: RefCell<Arc<Pager>>,
schema: RefCell<Arc<Schema>>,
/// Per-database schema cache (database_index -> schema)
/// Loaded lazily to avoid copying all schemas on connection open
@@ -1743,7 +1743,7 @@ impl Connection {
}
self.pager.borrow_mut().clear_page_cache();
let pager = self._db.init_pager(Some(size.get() as usize))?;
self.pager.replace(Rc::new(pager));
self.pager.replace(Arc::new(pager));
self.pager.borrow().set_initial_page_size(size);
Ok(())
@@ -1877,7 +1877,7 @@ impl Connection {
self._db.db_state.is_initialized()
}
fn get_pager_from_database_index(&self, index: &usize) -> Rc<Pager> {
fn get_pager_from_database_index(&self, index: &usize) -> Arc<Pager> {
if *index < 2 {
self.pager.borrow().clone()
} else {
@@ -1937,7 +1937,7 @@ impl Connection {
.with_views(use_views)
.with_strict(use_strict);
let db = Self::from_uri_attached(path, db_opts, self._db.io.clone())?;
let pager = Rc::new(db.init_pager(None)?);
let pager = Arc::new(db.init_pager(None)?);
self.attached_databases
.borrow_mut()
@@ -2092,7 +2092,7 @@ impl Connection {
databases
}
pub fn get_pager(&self) -> Rc<Pager> {
pub fn get_pager(&self) -> Arc<Pager> {
self.pager.borrow().clone()
}
@@ -2242,7 +2242,7 @@ pub struct Statement {
program: vdbe::Program,
state: vdbe::ProgramState,
mv_store: Option<Arc<MvStore>>,
pager: Rc<Pager>,
pager: Arc<Pager>,
/// Whether the statement accesses the database.
/// Used to determine whether we need to check for schema changes when
/// starting a transaction.
@@ -2259,7 +2259,7 @@ impl Statement {
pub fn new(
program: vdbe::Program,
mv_store: Option<Arc<MvStore>>,
pager: Rc<Pager>,
pager: Arc<Pager>,
query_mode: QueryMode,
) -> Self {
let accesses_db = program.accesses_db;

3
core/mvcc/cursor.rs
View File

@@ -5,7 +5,6 @@ use crate::Result;
use crate::{Pager, Value};
use std::fmt::Debug;
use std::ops::Bound;
use std::rc::Rc;
use std::sync::Arc;
#[derive(Debug, Copy, Clone)]
@@ -30,7 +29,7 @@ impl<Clock: LogicalClock> MvccLazyCursor<Clock> {
db: Arc<MvStore<Clock>>,
tx_id: u64,
table_id: u64,
pager: Rc<Pager>,
pager: Arc<Pager>,
) -> Result<MvccLazyCursor<Clock>> {
db.maybe_initialize_table(table_id, pager)?;
let cursor = Self {

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

@@ -25,7 +25,6 @@ use std::collections::HashSet;
use std::fmt::Debug;
use std::marker::PhantomData;
use std::ops::Bound;
use std::rc::Rc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use tracing::instrument;
@@ -303,7 +302,7 @@ struct CommitCoordinator {
pub struct CommitStateMachine<Clock: LogicalClock> {
state: CommitState,
is_finalized: bool,
pager: Rc<Pager>,
pager: Arc<Pager>,
tx_id: TxID,
connection: Arc<Connection>,
/// Write set sorted by table id and row id
@@ -351,7 +350,7 @@ pub struct DeleteRowStateMachine {
impl<Clock: LogicalClock> CommitStateMachine<Clock> {
fn new(
state: CommitState,
pager: Rc<Pager>,
pager: Arc<Pager>,
tx_id: TxID,
connection: Arc<Connection>,
commit_coordinator: Arc<CommitCoordinator>,
@@ -1306,7 +1305,7 @@ impl<Clock: LogicalClock> MvStore<Clock> {
/// to ensure exclusive write access as per SQLite semantics.
pub fn begin_exclusive_tx(
&self,
pager: Rc<Pager>,
pager: Arc<Pager>,
maybe_existing_tx_id: Option<TxID>,
) -> Result<IOResult<TxID>> {
self._begin_exclusive_tx(pager, false, maybe_existing_tx_id)
@@ -1318,7 +1317,7 @@ impl<Clock: LogicalClock> MvStore<Clock> {
/// to ensure exclusive write access as per SQLite semantics.
pub fn upgrade_to_exclusive_tx(
&self,
pager: Rc<Pager>,
pager: Arc<Pager>,
maybe_existing_tx_id: Option<TxID>,
) -> Result<IOResult<TxID>> {
self._begin_exclusive_tx(pager, true, maybe_existing_tx_id)
@@ -1331,7 +1330,7 @@ impl<Clock: LogicalClock> MvStore<Clock> {
#[instrument(skip_all, level = Level::DEBUG)]
fn _begin_exclusive_tx(
&self,
pager: Rc<Pager>,
pager: Arc<Pager>,
is_upgrade_from_read: bool,
maybe_existing_tx_id: Option<TxID>,
) -> Result<IOResult<TxID>> {
@@ -1390,7 +1389,7 @@ impl<Clock: LogicalClock> MvStore<Clock> {
/// This function starts a new transaction in the database and returns a `TxID` value
/// that you can use to perform operations within the transaction. All changes made within the
/// transaction are isolated from other transactions until you commit the transaction.
pub fn begin_tx(&self, pager: Rc<Pager>) -> Result<TxID> {
pub fn begin_tx(&self, pager: Arc<Pager>) -> Result<TxID> {
let tx_id = self.get_tx_id();
let begin_ts = self.get_timestamp();
let tx = Transaction::new(tx_id, begin_ts);
@@ -1418,7 +1417,7 @@ impl<Clock: LogicalClock> MvStore<Clock> {
pub fn commit_tx(
&self,
tx_id: TxID,
pager: Rc<Pager>,
pager: Arc<Pager>,
connection: &Arc<Connection>,
) -> Result<StateMachine<CommitStateMachine<Clock>>> {
let state_machine: StateMachine<CommitStateMachine<Clock>> =
@@ -1444,7 +1443,7 @@ impl<Clock: LogicalClock> MvStore<Clock> {
pub fn rollback_tx(
&self,
tx_id: TxID,
pager: Rc<Pager>,
pager: Arc<Pager>,
connection: &Connection,
) -> Result<()> {
let tx_unlocked = self.txs.get(&tx_id).unwrap();
@@ -1679,7 +1678,7 @@ impl<Clock: LogicalClock> MvStore<Clock> {
///
/// # Arguments
///
pub fn maybe_initialize_table(&self, table_id: u64, pager: Rc<Pager>) -> Result<()> {
pub fn maybe_initialize_table(&self, table_id: u64, pager: Arc<Pager>) -> Result<()> {
tracing::trace!("scan_row_ids_for_table(table_id={})", table_id);
// First, check if the table is already loaded.
@@ -1698,7 +1697,7 @@ impl<Clock: LogicalClock> MvStore<Clock> {
/// Scans the table and inserts the rows into the database.
///
/// This is initialization step for a table, where we still don't have any rows so we need to insert them if there are.
fn scan_load_table(&self, table_id: u64, pager: Rc<Pager>) -> Result<()> {
fn scan_load_table(&self, table_id: u64, pager: Arc<Pager>) -> Result<()> {
let root_page = table_id as usize;
let mut cursor = BTreeCursor::new_table(
None, // No MVCC cursor for scanning

4
core/schema.rs
View File

@@ -295,10 +295,10 @@ impl Schema {
pub fn make_from_btree(
&mut self,
mv_cursor: Option<Rc<RefCell<MvCursor>>>,
pager: Rc<Pager>,
pager: Arc<Pager>,
syms: &SymbolTable,
) -> Result<()> {
let mut cursor = BTreeCursor::new_table(mv_cursor, pager.clone(), 1, 10);
let mut cursor = BTreeCursor::new_table(mv_cursor, Arc::clone(&pager), 1, 10);
let mut from_sql_indexes = Vec::with_capacity(10);
let mut automatic_indices: HashMap<String, Vec<(String, usize)>> =

30
core/storage/btree.rs
View File

@@ -481,7 +481,7 @@ pub struct BTreeCursor {
/// The multi-version cursor that is used to read and write to the database file.
mv_cursor: Option<Rc<RefCell<MvCursor>>>,
/// The pager that is used to read and write to the database file.
pub pager: Rc<Pager>,
pub pager: Arc<Pager>,
/// Cached value of the usable space of a BTree page, since it is very expensive to call in a hot loop via pager.usable_space().
/// This is OK to cache because both 'PRAGMA page_size' and '.filectrl reserve_bytes' only have an effect on:
/// 1. an uninitialized database,
@@ -584,7 +584,7 @@ impl BTreeNodeState {
impl BTreeCursor {
pub fn new(
mv_cursor: Option<Rc<RefCell<MvCursor>>>,
pager: Rc<Pager>,
pager: Arc<Pager>,
root_page: usize,
num_columns: usize,
) -> Self {
@@ -635,7 +635,7 @@ impl BTreeCursor {
pub fn new_table(
mv_cursor: Option<Rc<RefCell<MvCursor>>>,
pager: Rc<Pager>,
pager: Arc<Pager>,
root_page: usize,
num_columns: usize,
) -> Self {
@@ -644,7 +644,7 @@ impl BTreeCursor {
pub fn new_index(
mv_cursor: Option<Rc<RefCell<MvCursor>>>,
pager: Rc<Pager>,
pager: Arc<Pager>,
root_page: usize,
index: &Index,
num_columns: usize,
@@ -5832,7 +5832,7 @@ impl std::fmt::Debug for IntegrityCheckState {
pub fn integrity_check(
state: &mut IntegrityCheckState,
errors: &mut Vec<IntegrityCheckError>,
pager: &Rc<Pager>,
pager: &Arc<Pager>,
) -> Result<IOResult<()>> {
loop {
let Some(IntegrityCheckPageEntry {
@@ -6108,7 +6108,7 @@ pub fn integrity_check(
}
pub fn btree_read_page(
pager: &Rc<Pager>,
pager: &Arc<Pager>,
page_idx: usize,
) -> Result<(Arc<Page>, Option<Completion>)> {
pager.read_page(page_idx)
@@ -7476,7 +7476,7 @@ fn fill_cell_payload(
cell_idx: usize,
record: &ImmutableRecord,
usable_space: usize,
pager: Rc<Pager>,
pager: Arc<Pager>,
fill_cell_payload_state: &mut FillCellPayloadState,
) -> Result<IOResult<()>> {
let overflow_page_pointer_size = 4;
@@ -7801,7 +7801,7 @@ mod tests {
fn insert_record(
cursor: &mut BTreeCursor,
pager: &Rc<Pager>,
pager: &Arc<Pager>,
rowid: i64,
val: Value,
) -> Result<(), LimboError> {
@@ -7888,7 +7888,7 @@ mod tests {
}
}
fn validate_btree(pager: Rc<Pager>, page_idx: usize) -> (usize, bool) {
fn validate_btree(pager: Arc<Pager>, page_idx: usize) -> (usize, bool) {
let num_columns = 5;
let cursor = BTreeCursor::new_table(None, pager.clone(), page_idx, num_columns);
let (page, _c) = cursor.read_page(page_idx).unwrap();
@@ -7998,7 +7998,7 @@ mod tests {
(depth.unwrap(), valid)
}
fn format_btree(pager: Rc<Pager>, page_idx: usize, depth: usize) -> String {
fn format_btree(pager: Arc<Pager>, page_idx: usize, depth: usize) -> String {
let num_columns = 5;
let cursor = BTreeCursor::new_table(None, pager.clone(), page_idx, num_columns);
@@ -8056,7 +8056,7 @@ mod tests {
}
}
fn empty_btree() -> (Rc<Pager>, usize, Arc<Database>, Arc<Connection>) {
fn empty_btree() -> (Arc<Pager>, usize, Arc<Database>, Arc<Connection>) {
#[allow(clippy::arc_with_non_send_sync)]
let io: Arc<dyn IO> = Arc::new(MemoryIO::new());
let db = Database::open_file(io.clone(), ":memory:", false, false).unwrap();
@@ -8739,7 +8739,7 @@ mod tests {
}
fn validate_expected_keys(
pager: &Rc<Pager>,
pager: &Arc<Pager>,
cursor: &mut BTreeCursor,
expected_keys: &[Vec<u8>],
seed: u64,
@@ -8967,7 +8967,7 @@ mod tests {
}
#[allow(clippy::arc_with_non_send_sync)]
fn setup_test_env(database_size: u32) -> Rc<Pager> {
fn setup_test_env(database_size: u32) -> Arc<Pager> {
let page_size = 512;
let io: Arc<dyn IO> = Arc::new(MemoryIO::new());
@@ -8985,7 +8985,7 @@ mod tests {
buffer_pool.clone(),
)));
let pager = Rc::new(
let pager = Arc::new(
Pager::new(
db_file,
Some(wal),
@@ -10487,7 +10487,7 @@ mod tests {
}
}
fn insert_cell(cell_idx: u64, size: u16, page: PageRef, pager: Rc<Pager>) {
fn insert_cell(cell_idx: u64, size: u16, page: PageRef, pager: Arc<Pager>) {
let mut payload = Vec::new();
let regs = &[Register::Value(Value::Blob(vec![0; size as usize]))];
let record = ImmutableRecord::from_registers(regs, regs.len());

3
core/translate/mod.rs
View File

@@ -53,7 +53,6 @@ use insert::translate_insert;
use rollback::translate_rollback;
use schema::{translate_create_table, translate_create_virtual_table, translate_drop_table};
use select::translate_select;
use std::rc::Rc;
use std::sync::Arc;
use tracing::{instrument, Level};
use transaction::{translate_tx_begin, translate_tx_commit};
@@ -65,7 +64,7 @@ use update::translate_update;
pub fn translate(
schema: &Schema,
stmt: ast::Stmt,
pager: Rc<Pager>,
pager: Arc<Pager>,
connection: Arc<Connection>,
syms: &SymbolTable,
query_mode: QueryMode,

11
core/translate/pragma.rs
View File

@@ -2,7 +2,6 @@
//! More info: https://www.sqlite.org/pragma.html.
use chrono::Datelike;
use std::rc::Rc;
use std::sync::Arc;
use turso_macros::match_ignore_ascii_case;
use turso_parser::ast::{self, ColumnDefinition, Expr, Literal, Name};
@@ -39,7 +38,7 @@ pub fn translate_pragma(
syms: &SymbolTable,
name: &ast::QualifiedName,
body: Option<ast::PragmaBody>,
pager: Rc<Pager>,
pager: Arc<Pager>,
connection: Arc<crate::Connection>,
mut program: ProgramBuilder,
) -> crate::Result<ProgramBuilder> {
@@ -90,7 +89,7 @@ fn update_pragma(
schema: &Schema,
syms: &SymbolTable,
value: ast::Expr,
pager: Rc<Pager>,
pager: Arc<Pager>,
connection: Arc<crate::Connection>,
mut program: ProgramBuilder,
) -> crate::Result<(ProgramBuilder, TransactionMode)> {
@@ -373,7 +372,7 @@ fn query_pragma(
pragma: PragmaName,
schema: &Schema,
value: Option<ast::Expr>,
pager: Rc<Pager>,
pager: Arc<Pager>,
connection: Arc<crate::Connection>,
mut program: ProgramBuilder,
) -> crate::Result<(ProgramBuilder, TransactionMode)> {
@@ -710,7 +709,7 @@ fn emit_columns_for_table_info(
fn update_auto_vacuum_mode(
auto_vacuum_mode: AutoVacuumMode,
largest_root_page_number: u32,
pager: Rc<Pager>,
pager: Arc<Pager>,
) -> crate::Result<()> {
pager.io.block(|| {
pager.with_header_mut(|header| {
@@ -723,7 +722,7 @@ fn update_auto_vacuum_mode(
fn update_cache_size(
value: i64,
pager: Rc<Pager>,
pager: Arc<Pager>,
connection: Arc<crate::Connection>,
) -> crate::Result<()> {
let mut cache_size_unformatted: i64 = value;

262
core/vdbe/execute.rs
View File

File diff suppressed because it is too large Load Diff

20
core/vdbe/mod.rs
View File

@@ -61,7 +61,7 @@ use execute::{
use explain::{insn_to_row_with_comment, EXPLAIN_COLUMNS, EXPLAIN_QUERY_PLAN_COLUMNS};
use regex::Regex;
use std::{cell::Cell, collections::HashMap, num::NonZero, rc::Rc, sync::Arc};
use std::{cell::Cell, collections::HashMap, num::NonZero, sync::Arc};
use tracing::{instrument, Level};
/// State machine for committing view deltas with I/O handling
@@ -492,7 +492,7 @@ pub struct Program {
}
impl Program {
fn get_pager_from_database_index(&self, idx: &usize) -> Rc<Pager> {
fn get_pager_from_database_index(&self, idx: &usize) -> Arc<Pager> {
self.connection.get_pager_from_database_index(idx)
}
@@ -500,7 +500,7 @@ impl Program {
&self,
state: &mut ProgramState,
mv_store: Option<Arc<MvStore>>,
pager: Rc<Pager>,
pager: Arc<Pager>,
query_mode: QueryMode,
) -> Result<StepResult> {
match query_mode {
@@ -514,7 +514,7 @@ impl Program {
&self,
state: &mut ProgramState,
_mv_store: Option<Arc<MvStore>>,
pager: Rc<Pager>,
pager: Arc<Pager>,
) -> Result<StepResult> {
debug_assert!(state.column_count() == EXPLAIN_COLUMNS.len());
if self.connection.closed.get() {
@@ -568,7 +568,7 @@ impl Program {
&self,
state: &mut ProgramState,
_mv_store: Option<Arc<MvStore>>,
pager: Rc<Pager>,
pager: Arc<Pager>,
) -> Result<StepResult> {
debug_assert!(state.column_count() == EXPLAIN_QUERY_PLAN_COLUMNS.len());
loop {
@@ -616,7 +616,7 @@ impl Program {
&self,
state: &mut ProgramState,
mv_store: Option<Arc<MvStore>>,
pager: Rc<Pager>,
pager: Arc<Pager>,
) -> Result<StepResult> {
let enable_tracing = tracing::enabled!(tracing::Level::TRACE);
loop {
@@ -692,7 +692,7 @@ impl Program {
&self,
state: &mut ProgramState,
rollback: bool,
pager: &Rc<Pager>,
pager: &Arc<Pager>,
) -> Result<IOResult<()>> {
use crate::types::IOResult;
@@ -792,7 +792,7 @@ impl Program {
pub fn commit_txn(
&self,
pager: Rc<Pager>,
pager: Arc<Pager>,
program_state: &mut ProgramState,
mv_store: Option<&Arc<MvStore>>,
rollback: bool,
@@ -895,7 +895,7 @@ impl Program {
#[instrument(skip(self, pager, connection), level = Level::DEBUG)]
fn step_end_write_txn(
&self,
pager: &Rc<Pager>,
pager: &Arc<Pager>,
commit_state: &mut CommitState,
connection: &Connection,
rollback: bool,
@@ -1061,7 +1061,7 @@ impl Row {
/// Handle a program error by rolling back the transaction
pub fn handle_program_error(
pager: &Rc<Pager>,
pager: &Arc<Pager>,
connection: &Connection,
err: &LimboError,
mv_store: Option<&Arc<MvStore>>,