在Rust中使用sqlx动态生成SQL时如何避免E0515错误？

问题场景 #

使用Rust结合sqlx和SQLite开发应用，动态生成SQL查询时遇到生命周期管理问题，简化代码如下：

use sqlx::{sqlite::SqlitePool, query_as, FromRow};
use sqlx::Error;
use futures::stream::Stream;
use futures::StreamExt;
use std::pin::Pin;

struct Dao {
    pool: SqlitePool,
}

impl Dao {
    pub fn new(pool: SqlitePool) -> Self {
        Self { pool }
    }

    pub async fn find<'a>(&'a self, order_by: &str) -> Result<Pin<Box<dyn Stream<Item = Result<MyStruct, Error>> + Send + 'a>>, Error> {
        let sql = format!("SELECT * FROM my_table ORDER BY {}", order_by);

        // Execute query
        let stream = query_as::<_, MyStruct>(&sql)
            .fetch(&self.pool);

        // Return the stream
        Ok(Box::pin(stream))
    }
}

#[derive(FromRow, Debug)]
struct MyStruct {
    id: i32,
    name: String,
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    let pool = SqlitePool::connect("sqlite::memory:").await?;

    // Create table
    sqlx::query("CREATE TABLE my_table (id INTEGER PRIMARY KEY, name TEXT)")
        .execute(&pool).await?;

    // Insert sample data
    sqlx::query("INSERT INTO my_table (name) VALUES ('example 1')")
        .execute(&pool).await?;
    sqlx::query("INSERT INTO my_table (name) VALUES ('example 2')")
        .execute(&pool).await?;

    let dao = Dao::new(pool);
    let order_by = "name DESC";
    let mut stream = dao.find(order_by).await?;

    while let Some(result) = stream.next().await {
        match result {
            Ok(record) => println!("Found record: {} - {}", record.id, record.name),
            Err(e) => eprintln!("Error: {}", e),
        }
    }

    Ok(())
}

运行时出现错误：

error[E0515]: cannot return value referencing local variable `sql`
  --> src/issues/avoid-e0515.rs:24:9
   |
20 |         let stream = query_as::<_, MyStruct>(&sql)
   |                                              ---- `sql` is borrowed here
...
24 |         Ok(Box::pin(stream))
   |         ^^^^^^^^^^^^^^^^^^^^ returns a value referencing data owned by the current function

核心原因 #

sqlx的query_as函数返回的QueryAs结构体持有传入的SQL字符串的引用，而动态生成的sql是函数内的本地变量，函数执行完毕后会被销毁。返回的Stream依赖该引用，导致生命周期不匹配，触发编译错误。

解决方案 #

方案1：一次性获取所有结果（简单无副作用） #

如果处理的数据量不大，直接使用fetch_all一次性获取所有查询结果，返回拥有所有权的Vec<MyStruct>，彻底避免流的生命周期问题：

impl Dao {
    // ... 其他方法不变

    pub async fn find(&self, order_by: &str) -> Result<Vec<MyStruct>, Error> {
        let sql = format!("SELECT * FROM my_table ORDER BY {}", order_by);
        let results = query_as::<_, MyStruct>(&sql)
            .fetch_all(&self.pool)
            .await?;
        Ok(results)
    }
}

// 调用方式修改为：
let records = dao.find(order_by).await?;
for record in records {
    println!("Found record: {} - {}", record.id, record.name);
}

方案2：使用async-stream创建自定义流（适合大数据量） #

需要流式处理结果时，使用async-stream crate创建自定义流，将SQL字符串的所有权捕获到流内部，避免引用失效：

1. 先在Cargo.toml添加依赖：

[dependencies]
async-stream = "0.3"

2. 修改find函数：

use async_stream::stream;

impl Dao {
    // ... 其他方法不变

    pub async fn find(&self, order_by: &str) -> Result<Pin<Box<dyn Stream<Item = Result<MyStruct, Error>> + Send>>, Error> {
        let sql = format!("SELECT * FROM my_table ORDER BY {}", order_by);
        let pool = self.pool.clone(); // 克隆连接池供流内部使用

        let custom_stream = stream! {
            // 在流内部执行查询，捕获sql的所有权
            let mut query_stream = query_as::<_, MyStruct>(&sql)
                .fetch(&pool);
            
            while let Some(item) = query_stream.next().await {
                yield item;
            }
        };

        Ok(Box::pin(custom_stream))
    }
}

方案3：使用Box::leak转为静态字符串（不推荐） #

通过将SQL字符串转为&'static str绕过生命周期检查，但会造成内存泄漏（字符串占用的内存永远不会被释放），仅适合调用次数极少的场景：

impl Dao {
    // ... 其他方法不变

    pub async fn find(&self, order_by: &str) -> Result<Pin<Box<dyn Stream<Item = Result<MyStruct, Error>> + Send + 'static>>, Error> {
        let sql = format!("SELECT * FROM my_table ORDER BY {}", order_by);
        // 将字符串转为静态引用，内存无法被释放
        let static_sql = Box::leak(sql.into_boxed_str());
        
        let stream = query_as::<_, MyStruct>(static_sql)
            .fetch(&self.pool);

        Ok(Box::pin(stream))
    }
}

额外注意事项 #

动态生成SQL时必须防范SQL注入风险，比如对order_by参数进行合法性校验，确保输入是允许的列名：

// 示例：验证排序字段是否合法
let allowed_columns = ["id", "name"];
let order_by_parts: Vec<&str> = order_by.split_whitespace().collect();
if let Some(col) = order_by_parts.first() {
    if !allowed_columns.contains(col) {
        return Err(Error::RowNotFound); // 可自定义错误类型
    }
}

内容的提问来源于stack exchange，提问作者Soichiro Shishido