igration tool to generate the correct next id (#1561)
This commit is contained in:
mdecimus
@@ -4,18 +4,21 @@
|
||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||
*/
|
||||
|
||||
use rusqlite::{OptionalExtension, TransactionBehavior, params};
|
||||
|
||||
use super::{SqliteStore, into_error};
|
||||
use crate::{
|
||||
IndexKey, Key, LogKey, SUBSPACE_COUNTER, SUBSPACE_IN_MEMORY_COUNTER, SUBSPACE_QUOTA, U64_LEN,
|
||||
write::{AssignedIds, Batch, BitmapClass, Operation, ValueClass, ValueOp},
|
||||
};
|
||||
|
||||
use super::{SqliteStore, into_error};
|
||||
use rusqlite::{OptionalExtension, TransactionBehavior, params};
|
||||
use trc::AddContext;
|
||||
|
||||
impl SqliteStore {
|
||||
pub(crate) async fn write(&self, batch: Batch<'_>) -> trc::Result<AssignedIds> {
|
||||
let mut conn = self.conn_pool.get().map_err(into_error)?;
|
||||
let mut conn = self
|
||||
.conn_pool
|
||||
.get()
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?;
|
||||
self.spawn_worker(move || {
|
||||
let mut account_id = u32::MAX;
|
||||
let mut collection = u8::MAX;
|
||||
@@ -23,7 +26,8 @@ impl SqliteStore {
|
||||
let mut change_id = 0u64;
|
||||
let trx = conn
|
||||
.transaction_with_behavior(TransactionBehavior::Immediate)
|
||||
.map_err(into_error)?;
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?;
|
||||
let mut result = AssignedIds::default();
|
||||
let has_changes = !batch.changes.is_empty();
|
||||
|
||||
@@ -36,9 +40,11 @@ impl SqliteStore {
|
||||
"ON CONFLICT(k) DO UPDATE SET v = v + ",
|
||||
"excluded.v RETURNING v"
|
||||
))
|
||||
.map_err(into_error)?
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?
|
||||
.query_row(params![&key, &1i64], |row| row.get::<_, i64>(0))
|
||||
.map_err(into_error)?;
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?;
|
||||
result.push_change_id(account_id, change_id as u64);
|
||||
}
|
||||
}
|
||||
@@ -81,9 +87,11 @@ impl SqliteStore {
|
||||
"INSERT OR REPLACE INTO {} (k, v) VALUES (?, ?)",
|
||||
table
|
||||
))
|
||||
.map_err(into_error)?
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?
|
||||
.execute([&key, value])
|
||||
.map_err(into_error)?;
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?;
|
||||
}
|
||||
ValueOp::AtomicAdd(by) => {
|
||||
if *by >= 0 {
|
||||
@@ -94,16 +102,20 @@ impl SqliteStore {
|
||||
),
|
||||
table
|
||||
))
|
||||
.map_err(into_error)?
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?
|
||||
.execute(params![&key, *by])
|
||||
.map_err(into_error)?;
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?;
|
||||
} else {
|
||||
trx.prepare_cached(&format!(
|
||||
"UPDATE {table} SET v = v + ? WHERE k = ?"
|
||||
))
|
||||
.map_err(into_error)?
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?
|
||||
.execute(params![*by, &key])
|
||||
.map_err(into_error)?;
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?;
|
||||
}
|
||||
}
|
||||
ValueOp::AddAndGet(by) => {
|
||||
@@ -116,16 +128,20 @@ impl SqliteStore {
|
||||
),
|
||||
table
|
||||
))
|
||||
.map_err(into_error)?
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?
|
||||
.query_row(params![&key, &*by], |row| row.get::<_, i64>(0))
|
||||
.map_err(into_error)?,
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?,
|
||||
);
|
||||
}
|
||||
ValueOp::Clear => {
|
||||
trx.prepare_cached(&format!("DELETE FROM {} WHERE k = ?", table))
|
||||
.map_err(into_error)?
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?
|
||||
.execute([&key])
|
||||
.map_err(into_error)?;
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -141,14 +157,18 @@ impl SqliteStore {
|
||||
|
||||
if *set {
|
||||
trx.prepare_cached("INSERT OR IGNORE INTO i (k) VALUES (?)")
|
||||
.map_err(into_error)?
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?
|
||||
.execute([&key])
|
||||
.map_err(into_error)?;
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?;
|
||||
} else {
|
||||
trx.prepare_cached("DELETE FROM i WHERE k = ?")
|
||||
.map_err(into_error)?
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?
|
||||
.execute([&key])
|
||||
.map_err(into_error)?;
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?;
|
||||
}
|
||||
}
|
||||
Operation::Bitmap { class, set } => {
|
||||
@@ -159,23 +179,29 @@ impl SqliteStore {
|
||||
if *set {
|
||||
if is_document_id {
|
||||
trx.prepare_cached("INSERT INTO b (k) VALUES (?)")
|
||||
.map_err(into_error)?
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?
|
||||
.execute(params![&key])
|
||||
.map_err(into_error)?;
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?;
|
||||
} else {
|
||||
trx.prepare_cached(&format!(
|
||||
"INSERT OR IGNORE INTO {} (k) VALUES (?)",
|
||||
table
|
||||
))
|
||||
.map_err(into_error)?
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?
|
||||
.execute(params![&key])
|
||||
.map_err(into_error)?;
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?;
|
||||
}
|
||||
} else {
|
||||
trx.prepare_cached(&format!("DELETE FROM {} WHERE k = ?", table))
|
||||
.map_err(into_error)?
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?
|
||||
.execute(params![&key])
|
||||
.map_err(into_error)?;
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?;
|
||||
};
|
||||
}
|
||||
Operation::Log { collection, set } => {
|
||||
@@ -187,9 +213,11 @@ impl SqliteStore {
|
||||
.serialize(0);
|
||||
|
||||
trx.prepare_cached("INSERT OR REPLACE INTO l (k, v) VALUES (?, ?)")
|
||||
.map_err(into_error)?
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?
|
||||
.execute([&key, set])
|
||||
.map_err(into_error)?;
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?;
|
||||
}
|
||||
Operation::AssertValue {
|
||||
class,
|
||||
@@ -200,16 +228,22 @@ impl SqliteStore {
|
||||
|
||||
let matches = trx
|
||||
.prepare_cached(&format!("SELECT v FROM {} WHERE k = ?", table))
|
||||
.map_err(into_error)?
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?
|
||||
.query_row([&key], |row| {
|
||||
Ok(assert_value.matches(row.get_ref(0)?.as_bytes()?))
|
||||
})
|
||||
.optional()
|
||||
.map_err(into_error)?
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?
|
||||
.unwrap_or_else(|| assert_value.is_none());
|
||||
if !matches {
|
||||
trx.rollback().map_err(into_error)?;
|
||||
return Err(trc::StoreEvent::AssertValueFailed.into());
|
||||
trx.rollback()
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?;
|
||||
return Err(trc::StoreEvent::AssertValueFailed
|
||||
.into_err()
|
||||
.caused_by(trc::location!()));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -221,13 +255,19 @@ impl SqliteStore {
|
||||
}
|
||||
|
||||
pub(crate) async fn purge_store(&self) -> trc::Result<()> {
|
||||
let conn = self.conn_pool.get().map_err(into_error)?;
|
||||
let conn = self
|
||||
.conn_pool
|
||||
.get()
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?;
|
||||
self.spawn_worker(move || {
|
||||
for subspace in [SUBSPACE_QUOTA, SUBSPACE_COUNTER, SUBSPACE_IN_MEMORY_COUNTER] {
|
||||
conn.prepare_cached(&format!("DELETE FROM {} WHERE v = 0", char::from(subspace),))
|
||||
.map_err(into_error)?
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?
|
||||
.execute([])
|
||||
.map_err(into_error)?;
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
@@ -236,15 +276,21 @@ impl SqliteStore {
|
||||
}
|
||||
|
||||
pub(crate) async fn delete_range(&self, from: impl Key, to: impl Key) -> trc::Result<()> {
|
||||
let conn = self.conn_pool.get().map_err(into_error)?;
|
||||
let conn = self
|
||||
.conn_pool
|
||||
.get()
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?;
|
||||
self.spawn_worker(move || {
|
||||
conn.prepare_cached(&format!(
|
||||
"DELETE FROM {} WHERE k >= ? AND k < ?",
|
||||
char::from(from.subspace()),
|
||||
))
|
||||
.map_err(into_error)?
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?
|
||||
.execute([from.serialize(0), to.serialize(0)])
|
||||
.map_err(into_error)?;
|
||||
.map_err(into_error)
|
||||
.caused_by(trc::location!())?;
|
||||
|
||||
Ok(())
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user