无法在循环内构建 rusqlite 事务：使用移动值并且一次不能多次借用可变值
Can't build a rusqlite transaction inside loop: use of moved value and cannot borrow as mutable more than once at a time
为了使用 rusqlite 加快插入 SQLite 数据库的速度，我想在 for 循环中构建一个事务并且只提交每 N 次迭代。
以下代码可以编译，但它构建了一个事务并一次性提交：
use rusqlite::{Connection, Result, NO_PARAMS};
fn main() -> Result<()> {
let mut conn = Connection::open_in_memory()?;
conn.execute(
"CREATE TABLE entry (
id INTEGER PRIMARY KEY,
data INTEGER
)",
NO_PARAMS,
)?;
let tx = conn.transaction()?;
for i in 0..20 {
tx.execute("INSERT INTO entry (data) VALUES (?1)", &[i])?;
}
tx.commit()?;
Ok(())
}
我的用例需要构建一个包含数百万个插入的事务，所以我想做的是在事务上累积，当它达到 transaction_size 时提交它并重新开始一个新的交易。 非编译版本看起来像这样：
let transaction_size = 5;
let tx = conn.transaction()?;
for i in 0..20 {
if (i % transaction_size) == (transaction_size - 1) {
tx.commit()?;
let tx = conn.transaction()?;
}
tx.execute("INSERT INTO entry (data) VALUES (?1)", &[i])?;
}
借阅检查器不允许这样做有两个原因。
error[E0382]: use of moved value: `tx`
--> src/main.rs:18:13
|
15 | let tx = conn.transaction()?;
| -- move occurs because `tx` has type `rusqlite::transaction::Transaction<'_>`, which does not implement the `Copy` trait
...
18 | tx.commit()?;
| ^^ value moved here, in previous iteration of loop
error[E0499]: cannot borrow `conn` as mutable more than once at a time
--> src/main.rs:19:22
|
15 | let tx = conn.transaction()?;
| ---- first mutable borrow occurs here
...
19 | let tx = conn.transaction()?;
| ^^^^ second mutable borrow occurs here
20 | }
21 | tx.execute("INSERT INTO entry (data) VALUES (?1)", &[i])?;
| -- first borrow later used here
第一个投诉对我来说很有意义。第二个不是那么多，因为下面会编译（但我每个事务只插入一行）：
for i in 0..20 {
let tx = conn.transaction()?;
tx.execute("INSERT INTO entry (data) VALUES (?1)", &[i])?;
tx.commit()?;
}
我试过在循环中使用 let tx = if cond { tx.commit()?; conn.transaction()? }，但你需要一个 else 子句来进行类型检查。
我不知道如何在让编译器满意的同时实现我的目标。也许有一些方法可以使用不安全的功能来做到这一点，但我对 Rust 还很陌生。
编辑
我忘了说我想把我的迭代器当作一次性使用。
使用来自@Sébastien Renauld 的将构建事务的逻辑分离到 do_batch 的想法，我制作了这个版本，它将累积必须使用可变向量添加到事务中的数据。然后它以 transaction_size.
大小的块构建并提交事务
use rusqlite::{Connection, Result, Transaction, NO_PARAMS};
use std::vec::Vec;
fn do_batch<'a>(tx: &Transaction<'a>, transaction_accum: &Vec<i32>) -> Result<()> {
for i in transaction_accum.iter() {
tx.execute("INSERT INTO entry (data) values (?1)", &[i])?;
}
Ok(())
}
fn main() -> Result<()> {
let mut conn = Connection::open_in_memory()?;
conn.execute(
"CREATE TABLE entry (
id INTEGER PRIMARY KEY,
data INTEGER
)",
NO_PARAMS,
)?;
let transaction_size = 5;
let mut transaction_accum: Vec<i32> = Vec::new();
for i in 1..20 {
transaction_accum.push(i);
if (i % transaction_size) == (transaction_size - 1) {
let tx = conn.transaction()?;
do_batch(&tx, &transaction_accum)?;
transaction_accum.clear();
tx.commit()?;
}
}
Ok(())
}
编辑 2
在@Sébastien Renauld 的另一个建议之后，我偶然发现了 itertools crate，它可以让你分块迭代器的输出，它提供了以下漂亮而干净的解决方案。我唯一担心的是，为了制作块，整个迭代器在调用 chunks 时在掩护下实现。是这样吗？
use rusqlite::{Connection, Result, Transaction, NO_PARAMS};
use std::vec::Vec;
use itertools::Itertools;
fn do_batch<'a>(tx: &Transaction<'a>, transaction_accum: &Vec<i32>) -> Result<()> {
for i in transaction_accum.iter() {
tx.execute("INSERT INTO entry (data) values (?1)", &[i])?;
}
Ok(())
}
fn main() -> Result<()> {
let mut conn = Connection::open_in_memory()?;
conn.execute(
"CREATE TABLE entry (
id INTEGER PRIMARY KEY,
data INTEGER
)",
NO_PARAMS,
)?;
let transaction_size = 5;
let my_iter = 1..20; // this is really a WalkDir from the walkdir crate
for chunk in &my_iter.into_iter().chunks(transaction_size) {
let tx = conn.transaction()?;
do_batch(&tx, &chunk.collect())?;
tx.commit()?;
}
Ok(())
}
这是一个 SQL 问题，而不是 Rust 问题，但我会解释你为什么 运行 参与其中，以及它如何出现在 Rust 中。
这一切都源于对事务数据库的基本误解，它适用于支持事务的每一个 RDBMS。事务的要点是打开服务器上可以看作是单独的平板的东西；然后，您对其进行状态更改，例如添加或删除行，然后 然后 您将单独的平板变成服务器的 "real" 状态。根据您使用的数据库引擎，这将以不同的方式实现，但对于我们今天针对您的问题的目的，这个类比就可以了。
您不是这样做，而是打开您的交易，进行一次插入，然后立即用 commit() 交回石板。注意它的签名：
fn commit(self) -> Result<()>
正如我们预期的那样，commit() 需要 self，而不是 &mut self。通过提交（或回滚），您告诉服务器您已完成此事务。
要解决此问题，您需要根据数据库决定如何处理它。批处理是一个好主意，您已经找到了，但是您需要确保能够承受一批失败并重复的后果。因此，我们将把事情分开一点。
首先，我们将构建我们的批处理生成器。我们将需要它，特别是如果我们打算重播一个批次：
fn do_batch<'a>(tx: &mut Transaction<'a>) -> Result<(), rusqlite::Error> {
for i in 0..20 {
tx.execute("INSERT INTO entry (data) values (?1", &[i])?;
}
Ok(())
}
然后，我们围绕它构建结构：
fn do_tx(mut conn: Connection) -> Result<(), rusqlite::Error> {
for i in 0..20 {
// Open the TX
let mut tx = conn.transaction()?;
do_batch(&mut tx)?;
// Do your error handling here. If the batch fails, you want to decide whether to retry or abort.
tx.commit()?;
}
Ok(())
}
如果可能，分离关注点总是值得的，如果需要，传递事务总是值得的；这就是他们在那里的目的。让您的函数构建批处理，然后在某种总体结构中处理 commit/rollback 行为。
正如你在评论中提到的，你正在走一棵树。为此，我假设您已经展平了迭代器（即您的 N 维树由一维迭代器表示），并且该迭代器位于 tree_walker.
目前没有在迭代器上定义 chunks() 方法，而这正是您需要的。为简洁起见，我们将 collect() 然后使用 Vec::chunks()。对于大多数工作负载，这应该不是问题，但如果您发现此分配的大小太大，您可以相对轻松地自己重新实现它。
use rusqlite::Error;
use rusqlite::{Connection, Transaction};
fn do_batch<'a>(tx: &Transaction<'a>, transaction_accum: &[i32]) -> Result<(), rusqlite::Error> {
for i in transaction_accum.iter() {
tx.execute("INSERT INTO entry (data) values (?1)", &[i])?;
}
Ok(())
}
fn commit(
mut conn: Connection,
tree_walker: impl Iterator<Item = i32>,
batch_size: usize,
) -> Result<(), rusqlite::Error> {
let collected: Vec<i32> = tree_walker.collect();
collected
.chunks(batch_size)
.fold(Ok(()), |current, elements| {
current.and_then(|_| {
let tx = conn.transaction()?;
do_batch(&tx, &elements)?;
tx.commit()
})
})
}
以下代码段的第 6 行存在一个重要的误解：
let transaction_size = 5;
let tx = conn.transaction()?;
for i in 0..20 {
if (i % transaction_size) == (transaction_size - 1) {
tx.commit()?;
let tx = conn.transaction()?; // <-- HERE
}
tx.execute("INSERT INTO entry (data) VALUES (?1)", &[i])?;
}
这一行并没有替换在第 2 行创建的 tx 变量，而是创建了一个名为 tx 的新变量，该变量在 [=14] 期间隐藏第一个变量=] 块并在它的末尾被丢弃。因此，当您到达 tx.execute 时，您返回尝试使用您已经提交的事务而不是新事务。
你想要的是：
let transaction_size = 5;
let mut tx = conn.transaction()?; // <-- Note the `mut` so that we can change it later to a new one
for i in 0..20 {
if (i % transaction_size) == (transaction_size - 1) {
tx.commit()?;
tx = conn.transaction()?; // <-- No `let` -> replace the existing `tx`
}
tx.execute("INSERT INTO entry (data) VALUES (?1)", &[i])?;
}
tx.commit()?; // <- Don't forget to commit the last transaction.
为了使用 rusqlite 加快插入 SQLite 数据库的速度，我想在 for 循环中构建一个事务并且只提交每 N 次迭代。
以下代码可以编译，但它构建了一个事务并一次性提交：
use rusqlite::{Connection, Result, NO_PARAMS};
fn main() -> Result<()> {
let mut conn = Connection::open_in_memory()?;
conn.execute(
"CREATE TABLE entry (
id INTEGER PRIMARY KEY,
data INTEGER
)",
NO_PARAMS,
)?;
let tx = conn.transaction()?;
for i in 0..20 {
tx.execute("INSERT INTO entry (data) VALUES (?1)", &[i])?;
}
tx.commit()?;
Ok(())
}
我的用例需要构建一个包含数百万个插入的事务，所以我想做的是在事务上累积，当它达到 transaction_size 时提交它并重新开始一个新的交易。 非编译版本看起来像这样：
let transaction_size = 5;
let tx = conn.transaction()?;
for i in 0..20 {
if (i % transaction_size) == (transaction_size - 1) {
tx.commit()?;
let tx = conn.transaction()?;
}
tx.execute("INSERT INTO entry (data) VALUES (?1)", &[i])?;
}
借阅检查器不允许这样做有两个原因。
error[E0382]: use of moved value: `tx`
--> src/main.rs:18:13
|
15 | let tx = conn.transaction()?;
| -- move occurs because `tx` has type `rusqlite::transaction::Transaction<'_>`, which does not implement the `Copy` trait
...
18 | tx.commit()?;
| ^^ value moved here, in previous iteration of loop
error[E0499]: cannot borrow `conn` as mutable more than once at a time
--> src/main.rs:19:22
|
15 | let tx = conn.transaction()?;
| ---- first mutable borrow occurs here
...
19 | let tx = conn.transaction()?;
| ^^^^ second mutable borrow occurs here
20 | }
21 | tx.execute("INSERT INTO entry (data) VALUES (?1)", &[i])?;
| -- first borrow later used here
第一个投诉对我来说很有意义。第二个不是那么多，因为下面会编译（但我每个事务只插入一行）：
for i in 0..20 {
let tx = conn.transaction()?;
tx.execute("INSERT INTO entry (data) VALUES (?1)", &[i])?;
tx.commit()?;
}
我试过在循环中使用 let tx = if cond { tx.commit()?; conn.transaction()? }，但你需要一个 else 子句来进行类型检查。
我不知道如何在让编译器满意的同时实现我的目标。也许有一些方法可以使用不安全的功能来做到这一点，但我对 Rust 还很陌生。
编辑
我忘了说我想把我的迭代器当作一次性使用。
使用来自@Sébastien Renauld 的将构建事务的逻辑分离到 do_batch 的想法，我制作了这个版本，它将累积必须使用可变向量添加到事务中的数据。然后它以 transaction_size.
use rusqlite::{Connection, Result, Transaction, NO_PARAMS};
use std::vec::Vec;
fn do_batch<'a>(tx: &Transaction<'a>, transaction_accum: &Vec<i32>) -> Result<()> {
for i in transaction_accum.iter() {
tx.execute("INSERT INTO entry (data) values (?1)", &[i])?;
}
Ok(())
}
fn main() -> Result<()> {
let mut conn = Connection::open_in_memory()?;
conn.execute(
"CREATE TABLE entry (
id INTEGER PRIMARY KEY,
data INTEGER
)",
NO_PARAMS,
)?;
let transaction_size = 5;
let mut transaction_accum: Vec<i32> = Vec::new();
for i in 1..20 {
transaction_accum.push(i);
if (i % transaction_size) == (transaction_size - 1) {
let tx = conn.transaction()?;
do_batch(&tx, &transaction_accum)?;
transaction_accum.clear();
tx.commit()?;
}
}
Ok(())
}
编辑 2
在@Sébastien Renauld 的另一个建议之后，我偶然发现了 itertools crate，它可以让你分块迭代器的输出，它提供了以下漂亮而干净的解决方案。我唯一担心的是，为了制作块，整个迭代器在调用 chunks 时在掩护下实现。是这样吗？
use rusqlite::{Connection, Result, Transaction, NO_PARAMS};
use std::vec::Vec;
use itertools::Itertools;
fn do_batch<'a>(tx: &Transaction<'a>, transaction_accum: &Vec<i32>) -> Result<()> {
for i in transaction_accum.iter() {
tx.execute("INSERT INTO entry (data) values (?1)", &[i])?;
}
Ok(())
}
fn main() -> Result<()> {
let mut conn = Connection::open_in_memory()?;
conn.execute(
"CREATE TABLE entry (
id INTEGER PRIMARY KEY,
data INTEGER
)",
NO_PARAMS,
)?;
let transaction_size = 5;
let my_iter = 1..20; // this is really a WalkDir from the walkdir crate
for chunk in &my_iter.into_iter().chunks(transaction_size) {
let tx = conn.transaction()?;
do_batch(&tx, &chunk.collect())?;
tx.commit()?;
}
Ok(())
}
这是一个 SQL 问题，而不是 Rust 问题，但我会解释你为什么 运行 参与其中，以及它如何出现在 Rust 中。
这一切都源于对事务数据库的基本误解，它适用于支持事务的每一个 RDBMS。事务的要点是打开服务器上可以看作是单独的平板的东西；然后，您对其进行状态更改，例如添加或删除行，然后 然后 您将单独的平板变成服务器的 "real" 状态。根据您使用的数据库引擎，这将以不同的方式实现，但对于我们今天针对您的问题的目的，这个类比就可以了。
您不是这样做，而是打开您的交易，进行一次插入，然后立即用 commit() 交回石板。注意它的签名：
fn commit(self) -> Result<()>
正如我们预期的那样，commit() 需要 self，而不是 &mut self。通过提交（或回滚），您告诉服务器您已完成此事务。
要解决此问题，您需要根据数据库决定如何处理它。批处理是一个好主意，您已经找到了，但是您需要确保能够承受一批失败并重复的后果。因此，我们将把事情分开一点。
首先，我们将构建我们的批处理生成器。我们将需要它，特别是如果我们打算重播一个批次：
fn do_batch<'a>(tx: &mut Transaction<'a>) -> Result<(), rusqlite::Error> {
for i in 0..20 {
tx.execute("INSERT INTO entry (data) values (?1", &[i])?;
}
Ok(())
}
然后，我们围绕它构建结构：
fn do_tx(mut conn: Connection) -> Result<(), rusqlite::Error> {
for i in 0..20 {
// Open the TX
let mut tx = conn.transaction()?;
do_batch(&mut tx)?;
// Do your error handling here. If the batch fails, you want to decide whether to retry or abort.
tx.commit()?;
}
Ok(())
}
如果可能，分离关注点总是值得的，如果需要，传递事务总是值得的；这就是他们在那里的目的。让您的函数构建批处理，然后在某种总体结构中处理 commit/rollback 行为。
正如你在评论中提到的，你正在走一棵树。为此，我假设您已经展平了迭代器（即您的 N 维树由一维迭代器表示），并且该迭代器位于 tree_walker.
目前没有在迭代器上定义 chunks() 方法，而这正是您需要的。为简洁起见，我们将 collect() 然后使用 Vec::chunks()。对于大多数工作负载，这应该不是问题，但如果您发现此分配的大小太大，您可以相对轻松地自己重新实现它。
use rusqlite::Error;
use rusqlite::{Connection, Transaction};
fn do_batch<'a>(tx: &Transaction<'a>, transaction_accum: &[i32]) -> Result<(), rusqlite::Error> {
for i in transaction_accum.iter() {
tx.execute("INSERT INTO entry (data) values (?1)", &[i])?;
}
Ok(())
}
fn commit(
mut conn: Connection,
tree_walker: impl Iterator<Item = i32>,
batch_size: usize,
) -> Result<(), rusqlite::Error> {
let collected: Vec<i32> = tree_walker.collect();
collected
.chunks(batch_size)
.fold(Ok(()), |current, elements| {
current.and_then(|_| {
let tx = conn.transaction()?;
do_batch(&tx, &elements)?;
tx.commit()
})
})
}
以下代码段的第 6 行存在一个重要的误解：
let transaction_size = 5;
let tx = conn.transaction()?;
for i in 0..20 {
if (i % transaction_size) == (transaction_size - 1) {
tx.commit()?;
let tx = conn.transaction()?; // <-- HERE
}
tx.execute("INSERT INTO entry (data) VALUES (?1)", &[i])?;
}
这一行并没有替换在第 2 行创建的 tx 变量，而是创建了一个名为 tx 的新变量，该变量在 [=14] 期间隐藏第一个变量=] 块并在它的末尾被丢弃。因此，当您到达 tx.execute 时，您返回尝试使用您已经提交的事务而不是新事务。
你想要的是：
let transaction_size = 5;
let mut tx = conn.transaction()?; // <-- Note the `mut` so that we can change it later to a new one
for i in 0..20 {
if (i % transaction_size) == (transaction_size - 1) {
tx.commit()?;
tx = conn.transaction()?; // <-- No `let` -> replace the existing `tx`
}
tx.execute("INSERT INTO entry (data) VALUES (?1)", &[i])?;
}
tx.commit()?; // <- Don't forget to commit the last transaction.