filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

documentId generation within the database transaction

Browse Source
This commit is contained in:
mdecimus
2024-04-23 20:03:30 +02:00
parent 03332635f8
commit c6960bfba8
74 changed files with 2132 additions and 3111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

129
crates/store/src/backend/sqlite/write.rs
View File

@@ -21,24 +21,28 @@
* for more details.
*/
use roaring::RoaringBitmap;
use rusqlite::{params, OptionalExtension, TransactionBehavior};
use crate::{
write::{Batch, BitmapClass, Operation, ValueClass, ValueOp},
BitmapKey, IndexKey, Key, LogKey, ValueKey, SUBSPACE_COUNTERS,
write::{
key::DeserializeBigEndian, AssignedIds, Batch, BitmapClass, Operation, RandomAvailableId,
ValueOp,
},
BitmapKey, IndexKey, Key, LogKey, SUBSPACE_COUNTERS, U32_LEN,
};
use super::SqliteStore;
impl SqliteStore {
pub(crate) async fn write(&self, batch: Batch) -> crate::Result<Option<i64>> {
pub(crate) async fn write(&self, batch: Batch) -> crate::Result<AssignedIds> {
let mut conn = self.conn_pool.get()?;
self.spawn_worker(move || {
let mut account_id = u32::MAX;
let mut collection = u8::MAX;
let mut document_id = u32::MAX;
let trx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
let mut result = None;
let mut result = AssignedIds::default();
for op in &batch.ops {
match op {
@@ -58,14 +62,14 @@ impl SqliteStore {
document_id = *document_id_;
}
Operation::Value { class, op } => {
let key = ValueKey {
let key = class.serialize(
account_id,
collection,
document_id,
class,
};
let table = char::from(key.subspace());
let key = key.serialize(0);
0,
(&result).into(),
);
let table = char::from(class.subspace(collection));
match op {
ValueOp::Set(value) => {
@@ -73,27 +77,7 @@ impl SqliteStore {
"INSERT OR REPLACE INTO {} (k, v) VALUES (?, ?)",
table
))?
.execute([&key, value])?;
if matches!(class, ValueClass::ReservedId) {
// Make sure the reserved id is not already in use
let key = BitmapKey {
account_id,
collection,
class: BitmapClass::DocumentIds,
block_num: document_id,
}
.serialize(0);
if trx
.prepare_cached("SELECT 1 FROM b WHERE k = ?")?
.query_row([&key], |_| Ok(true))
.optional()?
.unwrap_or(false)
{
trx.rollback()?;
return Err(crate::Error::AssertValueFailed);
}
}
.execute([&key, value.resolve(&result)?.as_ref()])?;
}
ValueOp::AtomicAdd(by) => {
if *by >= 0 {
@@ -108,13 +92,14 @@ impl SqliteStore {
}
}
ValueOp::AddAndGet(by) => {
result = trx
.prepare_cached(concat!(
result.push_counter_id(
trx.prepare_cached(concat!(
"INSERT INTO c (k, v) VALUES (?, ?) ",
"ON CONFLICT(k) DO UPDATE SET v = v + excluded.v RETURNING v"
"ON CONFLICT(k) DO UPDATE SET v = v + ",
"excluded.v RETURNING v"
))?
.query_row(params![&key, &by], |row| row.get::<_, i64>(0))?
.into();
.query_row(params![&key, &by], |row| row.get::<_, i64>(0))?,
);
}
ValueOp::Clear => {
trx.prepare_cached(&format!("DELETE FROM {} WHERE k = ?", table))?
@@ -141,49 +126,83 @@ impl SqliteStore {
}
}
Operation::Bitmap { class, set } => {
let key = BitmapKey {
// Find the next available document id
let is_document_id = matches!(class, BitmapClass::DocumentIds);
if *set && is_document_id && document_id == u32::MAX {
let begin = BitmapKey {
account_id,
collection,
class: BitmapClass::DocumentIds,
document_id: 0,
}
.serialize(0);
let end = BitmapKey {
account_id,
collection,
class: BitmapClass::DocumentIds,
document_id: u32::MAX,
}
.serialize(0);
let key_len = begin.len();
let mut query =
trx.prepare_cached("SELECT k FROM b WHERE k >= ? AND k <= ?")?;
let mut rows = query.query([&begin, &end])?;
let mut found_ids = RoaringBitmap::new();
while let Some(row) = rows.next()? {
let key = row.get_ref(0)?.as_bytes()?;
if key.len() == key_len {
found_ids.insert(key.deserialize_be_u32(key.len() - U32_LEN)?);
}
}
document_id = found_ids.random_available_id();
result.push_document_id(document_id);
}
let key = class.serialize(
account_id,
collection,
class,
block_num: document_id,
}
.serialize(0);
document_id,
0,
(&result).into(),
);
if *set {
trx.prepare_cached("INSERT OR IGNORE INTO b (k) VALUES (?)")?
.execute(params![&key])?;
if is_document_id {
trx.prepare_cached("INSERT INTO b (k) VALUES (?)")?
.execute(params![&key])?;
} else {
trx.prepare_cached("INSERT OR IGNORE INTO b (k) VALUES (?)")?
.execute(params![&key])?;
}
} else {
trx.prepare_cached("DELETE FROM b WHERE k = ?")?
.execute(params![&key])?;
};
}
Operation::Log {
collection,
change_id,
set,
} => {
Operation::Log { set } => {
let key = LogKey {
account_id,
collection: *collection,
change_id: *change_id,
collection,
change_id: batch.change_id,
}
.serialize(0);
trx.prepare_cached("INSERT OR REPLACE INTO l (k, v) VALUES (?, ?)")?
.execute([&key, set])?;
.execute([&key, set.resolve(&result)?.as_ref()])?;
}
Operation::AssertValue {
class,
assert_value,
} => {
let key = ValueKey {
let key = class.serialize(
account_id,
collection,
document_id,
class,
};
let table = char::from(key.subspace());
let key = key.serialize(0);
0,
(&result).into(),
);
let table = char::from(class.subspace(collection));
let matches = trx
.prepare_cached(&format!("SELECT v FROM {} WHERE k = ?", table))?