Rust in Everything
基于 sqlx 0.9.0

扩展查询协议与语句缓存

主线位置：查询主线的中段，&mut PgConnection 的 fetch_many / fetch_optional 调进来的 PgConnection::run。一次查询的全过程已经抓过第一次和第二次执行的消息（第一次 Parse + Describe + Sync 一个来回，再 Bind + Execute + Close + Sync；第二次命中缓存只剩后一组），也讲了 Bind 的二进制格式和 Execute 的 limit: 0。这一页不重复那些，讲 run 背后的两本账：欠了服务端几个 ReadyForQuery，以及每条连接上缓存了哪些预备语句。

flowchart TB
  start["run(sql, args)"] --> wur1[wait_until_ready]
  wur1 --> hasargs{"有 Arguments?"}
  hasargs -->|否| q["Query 消息"]
  hasargs -->|是| gop[get_or_prepare]
  gop --> hit{"缓存命中?"}
  hit -->|否| prep["Parse Describe Sync"]
  prep --> evict{"挤掉一条?"}
  evict -->|是| closes["Close S 加 Sync"]
  evict -->|否| patch[apply_patches]
  closes --> patch
  hit -->|是| patch
  patch --> wur2[wait_until_ready]
  wur2 --> bind["Bind Execute Close Sync"]
  bind --> fl[flush]
  q --> fl
  fl --> stream["try_stream 读到 ReadyForQuery"]

实验都经过连接选项与建立连接那页的代理（via_proxy），连接用 PgConnection::connect_with 直接建，不经过连接池，所以没有借出和归还时的 ping。平台：M2 Max、macOS、rustc 1.99.0、Postgres 16.14。

第一本账：pending_ready_for_query_count

协议里每个 Sync 换来一个 ReadyForQuery，每个简单查询的 Query 消息也换来一个。sqlx 在连接上记一个数：发出去的、还没收到回应的 ReadyForQuery 有几个。加一只在两处，都是「把消息写进缓冲」的那一刻：

    #[inline(always)]
    pub(crate) fn write_sync(&mut self) {
        self.inner
            .stream
            .write_msg(message::Sync)
            .expect("BUG: Sync should not be too big for protocol");

        // all SYNC messages will return a ReadyForQuery
        self.inner.pending_ready_for_query_count += 1;
    }

queue_simple_query 同理，写一条 Query 消息、计数加一。它有三个调用者：run 的简单查询分支、Transaction 被 drop 时排进去的 ROLLBACK（start_rollback）、PgAdvisoryLockGuard 被 drop 时排进去的解锁语句。后两者只写缓冲，不 flush，因为 Drop 里不能 .await。

减一在 handle_ready_for_query 和 recv_ready_for_query 里，前者用 checked_sub，收到的比欠的多就报 received more ReadyForQuery messages than expected（0.8.1 安全审计时加的，见消息与读写缓冲的冷知识）。收账的是 wait_until_ready：

    // will return when the connection is ready for another query
    pub(crate) async fn wait_until_ready(&mut self) -> Result<(), Error> {
        if !self.inner.stream.write_buffer_mut().is_empty() {
            self.inner.stream.flush().await?;
        }

        while self.inner.pending_ready_for_query_count > 0 {
            let message = self.inner.stream.recv().await?;

            if let BackendMessageFormat::ReadyForQuery = message.format {
                self.handle_ready_for_query(message)?;
            }
        }

        Ok(())
    }

先把写缓冲里攒着的东西发出去（比如排队的 ROLLBACK），再一条条读，除了 ReadyForQuery 什么都不看，直到账清零。run 开头调一次，prepare 里发 Parse 之前调一次，apply_patches 之后、写 Bind 之前又调一次（apply_patches 可能为自定义类型查过 OID，见 Postgres 类型）。ping 就是 write_sync 加 wait_until_ready。

正常跑完的查询不会留账：run 返回的 stream 读到 ReadyForQuery 时自己 handle_ready_for_query，然后结束。留账的情况都是「没读到最后」：

情况线上留下了什么谁来收
服务端回了 ErrorResponse错误之后的 ReadyForQuery下一次 wait_until_ready（消息与读写缓冲有抓包）
fetch 的 stream 读了几行就被 drop剩下所有的 DataRow、CommandComplete、ReadyForQuery同上
查询的 future 在 .await 中被取消（超时、select!）视取消点而定，可能连 flush 都没做完同上；但取消 begin 这类会改连接状态的调用另有问题，比如 #4423 报告的「cancelling Pool::begin_with can return an open transaction to the pool」
Transaction 没提交就 drop缓冲里一条 ROLLBACK，还没发同上，先 flush 再读

实验：stream 读三行就 drop

一条会返回 20 万行的查询，只读三行就把 stream 丢掉，等 500 毫秒，再跑下一条查询。代理里加了一个计数器，记录服务端到客户端一共转发了多少字节（src/bin/drop_stream.rs）：

#[tokio::main]
async fn main() -> anyhow::Result<()> {
  let (p, opts) = via_proxy().await?;
  quiet(&p, true);
  let mut conn = PgConnection::connect_with(&opts).await?;
  let bytes = || p.s2c_bytes.load(Ordering::SeqCst);
  let before = bytes();

  // 200_000 int4 rows; each DataRow is 15 bytes on the wire
  let mut s = sqlx::query_scalar::<_, i32>("select g from generate_series(1, $1) g")
    .bind(200_000_i32)
    .fetch(&mut conn);
  let mut got = Vec::new();
  for _ in 0..3 {
    got.push(s.try_next().await?.unwrap());
  }
  drop(s);
  println!("read {got:?}, dropped the stream");
  tokio::time::sleep(Duration::from_millis(500)).await;
  println!("500ms later: server->client bytes so far = {}", bytes() - before);

  quiet(&p, false);
  let t = Instant::now();
  let v: i32 = sqlx::query_scalar("select 42").fetch_one(&mut conn).await?;
  let elapsed = t.elapsed();
  quiet(&p, true);
  println!("next query -> {v} in {elapsed:?}; server->client bytes now = {}", bytes() - before);
  settle().await;
  Ok(())
}

代理把一次读里连续的同类消息合成一项，×N 是条数：

read [1, 2, 3], dropped the stream
500ms later: server->client bytes so far = 491569
  [conn 1] S->C DataRow(列长 4) ×4369
  [conn 1] S->C DataRow(列长 4) ×4369
  [conn 1] S->C DataRow(列长 4) ×4369
...
  [conn 1] S->C DataRow(列长 4) ×1208, CommandComplete("SELECT 200000"), CloseComplete, ReadyForQuery(I)
  [conn 1] C->S Parse(sqlx_s_2, "select 42"), Describe(S, sqlx_s_2), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_2), Execute(limit 0), Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
next query -> 42 in 183.858375ms; server->client bytes now = 3000181

（中间省略了 35 行，每行是 4369 或 4370 条 DataRow(列长 4)。）

  • drop 之后，服务端并没有停：sqlx 没有 CancelRequest 的代码（连接选项与建立连接里说过，BackendKeyData 存了但没用），服务端照常往外发。
  • 但它发不动了。客户端不读，内核的 socket 缓冲很快写满，TCP 背压一路传回服务端。500 毫秒里只转发了约 49 万字节（每行 15 字节，三万多行），服务端的发送就卡在那里。
  • 下一条 select 42 的 run 一进门就 wait_until_ready：账上欠一个 ReadyForQuery，于是把剩下的 16 万多行一条条 recv 进来扔掉，直到 ReadyForQuery。全部 300 万字节读完，才轮到 Parse(sqlx_s_2, ...)。这条只返回一行的查询花了 183 毫秒（debug 构建），几乎全花在收拾上一条查询上。
📝NOTE
没有蠢问题

问： 我只想要前几行，fetch 读够了就 drop，不是比 fetch_all 省吗？

答： 只省了解码，没省传输。服务端还是会把整个结果集算出来、发出来，客户端还是会把它们全部读进来，只是推迟到了这条连接下一次被使用的时候，由一条无关的查询（或者连接池归还时的 ping）来付这笔时间。要少拿行，在 SQL 里写 LIMIT，或者用服务端游标（DECLARE ... CURSOR 加 FETCH n）。sqlx 的 Execute 消息固定发 limit: 0，不会替你在协议层截断（原因见一次查询的全过程的冷知识）。

收账时用的是普通的 recv，它遇到 ErrorResponse 照样返回 Err。如果被丢掉的那部分结果里有错误，这个错误会由下一条查询报出来（src/bin/drop_then_error.rs）：

  // take the first row (10 / 2 = 5), drop the stream; row 3 divides by zero
  let mut s = sqlx::query_scalar::<_, i32>("select 10 / (3 - g) from generate_series(1, $1) g")
    .bind(3_i32)
    .fetch(&mut conn);
  let first = s.try_next().await?;
  drop(s);
  settle().await;
  println!("first row: {first:?}, stream dropped");

  for label in ["second query", "third query"] {
    let r = sqlx::query_scalar::<_, i32>("select 42").fetch_one(&mut conn).await;
    settle().await;
    println!("{label}: {r:?}");
  }

  [conn 1] C->S Parse(sqlx_s_1, "select 10 / (3 - g) from generate_series…"), Describe(S, sqlx_s_1), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute(limit 0), Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4), DataRow(列长 4), ErrorResponse(S=ERROR C=22012 M=division by zero)
  [conn 1] S->C ReadyForQuery(I)
first row: Some(5), stream dropped
second query: Err(Database(PgDatabaseError { severity: Error, code: "22012", message: "division by zero", detail: None, hint: None, position: None, where: None, schema: None, table: None, column: None, data_type: None, constraint: None, file: Some("int.c"), line: Some(870), routine: Some("int4div") }))
  [conn 1] C->S Parse(sqlx_s_2, "select 42"), Describe(S, sqlx_s_2), Sync
  ...
third query: Ok(42)

select 42 第一次执行时一个字节都没发，直接返回了上一条查询的「division by zero」：wait_until_ready 读到 ErrorResponse，? 把错误原样抛出，计数没有减，那个 ReadyForQuery 还在缓冲里。第三次调用时 wait_until_ready 才读到它，账清零，查询正常执行。没有找到对应的上游 issue。

经过连接池时，这一步落在归还上：stream 里的 PoolConnection 被 drop，return_to_pool 的 ping 拿到这个错误，于是 close_hard 掉这条连接。同样的程序换成 max_connections(1)、test_before_acquire(false) 的池（src/bin/drop_then_error_pool.rs），drop 之后 pool.size() 是 0，下一条查询新建了一条连接（抓包里出现了 [conn 2] 的完整握手），没有报错。归还的细节见归还与维护。

第二本账：StatementCache

每条连接一个 StatementCache，里面是 hashlink 的 LruCache<String, (StatementId, Arc<PgStatementMetadata>)>。键是 SQL 原文，一个字符都不能差：多一个空格就是另一条语句。get_or_prepare 先 get_mut(sql)，命中时 LruCache 顺便把这一项挪到「最近使用」的一端。没命中就 prepare，然后 insert：

    pub fn insert(&mut self, k: &str, v: T) -> Option<T> {
        let mut lru_item = None;

        if self.capacity() == self.len() && !self.contains_key(k) {
            lru_item = self.remove_lru();
        } else if self.contains_key(k) {
            lru_item = self.inner.remove(k);
        }

        self.inner.insert(k.into(), v);

        lru_item
    }

满了就挤掉最久没用的那一项，把它还给调用者。get_or_prepare 拿到被挤掉的语句 id，立刻让服务端也释放它：

        if persistent && self.inner.cache_statement.is_enabled() {
            if let Some((id, _)) = self.inner.cache_statement.insert(sql, statement.clone()) {
                self.inner.stream.write_msg(Close::Statement(id))?;
                self.write_sync();

                self.inner.stream.flush().await?;

                self.wait_for_close_complete(1).await?;
                self.recv_ready_for_query().await?;
            }
        }

这是一个单独的来回：Close 加 Sync，flush，等 CloseComplete 和 ReadyForQuery，然后才回到 run 去写 Bind。一次「没命中并且挤掉一条」的查询因此是三个来回。客户端缓存和服务端的预备语句始终一一对应，服务端不会攒下客户端已经忘掉的语句。

容量默认 100，用 PgConnectOptions::statement_cache_capacity 或 URL 的 statement-cache-capacity 改。整个缓存可以用 clear_cached_statements 清掉：它把每一项的 Close::Statement 都写进缓冲，只跟一个 Sync，一个来回关掉全部。

语句名是 sqlx_s_ 加一个序号。序号是 Option<NonZeroU32>，None 表示未命名语句，next() 用 checked_add(1)，到 u32::MAX 之后绕回 1。序号在发 Parse 之前就分配了，Parse 失败也不收回（消息与读写缓冲的实验里 sqlx_s_2 就是这样跳过的）。

❗IMPORTANT
动脑筋：容量为 2，四次查询之后服务端还剩哪两条？

statement_cache_capacity(2)，依次执行 select 1、select 2、select 1、select 3，最后查 pg_prepared_statements（src/bin/stmt_cache.rs）：

async fn names(conn: &mut PgConnection) -> anyhow::Result<Vec<String>> {
  // persistent(false): this statement itself must not enter the cache
  Ok(
    sqlx::query_scalar("select name from pg_prepared_statements order by name")
      .persistent(false)
      .fetch_all(conn)
      .await?,
  )
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
  let (p, opts) = via_proxy().await?;
  quiet(&p, true);
  let mut conn = PgConnection::connect_with(&opts.statement_cache_capacity(2)).await?;
  quiet(&p, false);

  for (label, sql) in [("q1", "select 1"), ("q2", "select 2"), ("q1 again", "select 1"), ("q3", "select 3")] {
    println!("--- {label} ---");
    let _: i32 = sqlx::query_scalar(sql).fetch_one(&mut conn).await?;
    settle().await;
  }
  quiet(&p, true);
  println!("cached = {}, server has {:?}", conn.cached_statements_size(), names(&mut conn).await?);
  Ok(())
}

select 3 进来时，被 Close 掉的是 sqlx_s_1 还是 sqlx_s_2？

persistent(false)：未命名语句

Query::persistent(false) 让这条查询不进缓存。prepare 的第一件事就是按它选语句名：

    let id = if persistent {
        let id = conn.inner.next_statement_id;
        conn.inner.next_statement_id = id.next();
        id
    } else {
        StatementId::UNNAMED
    };

未命名语句（名字是空串）是协议里的一个特殊槽位：每个会话只有一个，下一次 Parse 到未命名语句时旧的自动作废，一条简单查询的 Query 消息也会销毁它。所以它不需要 Close，也不会在服务端越攒越多。

📝NOTE
冷知识：persistent(false) 曾经会泄漏预备语句

0.8.6 之前，prepare 不管 persistent 是什么，都分配一个 sqlx_s_N；persistent(false) 只是让结果不进客户端缓存。issue #3850（「Non-persistent prepared statements not closed in PostgreSQL」）指出文档写的是「If false, the prepared statement will be closed after execution」，实际是「the prepared statement never gets closed」，同一条查询反复执行就在服务端反复建新的命名语句，「This can lead to a memory leak in the database」。PR #3863（「Use unnamed statement in pg when not persistent」，首次发布于 0.8.6，0.9.0 的 CHANGELOG 里也列了它）加了上面这个 if persistent，提交说明是「This will automatically close the prepared statement when another query is run, avoiding a memory leak」。

src/bin/unnamed.rs 的前半段，同一条 SQL 先 persistent(false) 跑两次，再默认跑一次，再 persistent(false) 跑一次：

  for round in 1..=2 {
    println!("--- persistent(false), round {round} ---");
    let _: i32 = sqlx::query_scalar("select $1::int4 + 1").bind(1_i32).persistent(false).fetch_one(&mut conn).await?;
    settle().await;
  }
  println!("--- persistent(true), same SQL ---");
  let _: i32 = sqlx::query_scalar("select $1::int4 + 1").bind(1_i32).fetch_one(&mut conn).await?;
  settle().await;
  println!("--- persistent(false) again, SQL now cached ---");
  let _: i32 = sqlx::query_scalar("select $1::int4 + 1").bind(1_i32).persistent(false).fetch_one(&mut conn).await?;
  settle().await;
  quiet(&p, true);
  println!("cached = {}, server has {:?}", conn.cached_statements_size(), names(&mut conn).await?);

--- persistent(false), round 1 ---
  [conn 1] C->S Parse(unnamed, "select $1::int4 + 1"), Describe(S, unnamed), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt unnamed), Execute(limit 0), Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
--- persistent(false), round 2 ---
  [conn 1] C->S Parse(unnamed, "select $1::int4 + 1"), Describe(S, unnamed), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt unnamed), Execute(limit 0), Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
--- persistent(true), same SQL ---
  [conn 1] C->S Parse(sqlx_s_1, "select $1::int4 + 1"), Describe(S, sqlx_s_1), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute(limit 0), Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
--- persistent(false) again, SQL now cached ---
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute(limit 0), Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
cached = 1, server has ["sqlx_s_1"]
  • persistent(false) 每次都是两个来回：Parse 到未命名语句、Describe、Sync，再 Bind 到 stmt unnamed。第二次和第一次一模一样，没有任何缓存。prepare 仍然是单独一个来回，因为 sqlx 要先拿到参数和列的描述，才能写 Bind、解码结果。
  • 默认的 persistent(true) 建了 sqlx_s_1 并进了缓存。
  • 最后一段：同一条 SQL 再用 persistent(false)，结果直接 Bind 到 sqlx_s_1，没有 Parse。get_or_prepare 先查缓存，命中就返回，根本走不到看 persistent 的那一步。persistent(false) 的准确含义是「这次执行的结果不写进缓存」，不是「不用缓存里的语句」。Query::persistent 的文档写了这一点：「If queried once with the flag set to true, all subsequent queries matching the one with the flag will use the cached statement until the cache is cleared」。

statement_cache_capacity(0)：0.9.0 里还漏着

把容量设成 0，本意是「不要语句缓存」。get_or_prepare 里写缓存的条件是 persistent && is_enabled()，容量 0 时 is_enabled() 为假，不写缓存，也就不会有挤出和 Close。可 prepare 选语句名只看 persistent。默认 persistent 是 true，于是每次执行都建一个新的命名语句，没人关它。src/bin/unnamed.rs 的中段：

  // statement_cache_capacity(0)
  let mut conn0 = PgConnection::connect_with(&opts.clone().statement_cache_capacity(0)).await?;
  quiet(&p, false);
  for round in 1..=2 {
    println!("--- capacity 0, round {round} ---");
    let _: i32 = sqlx::query_scalar("select $1::int4 * 2").bind(1_i32).fetch_one(&mut conn0).await?;
    settle().await;
  }
  quiet(&p, true);
  println!("capacity 0: cached = {}, server has {:?}", conn0.cached_statements_size(), names(&mut conn0).await?);

--- capacity 0, round 1 ---
  [conn 2] C->S Parse(sqlx_s_1, "select $1::int4 * 2"), Describe(S, sqlx_s_1), Sync
  [conn 2] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
  [conn 2] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute(limit 0), Close(P, unnamed), Sync
  [conn 2] S->C BindComplete, DataRow(列长 4), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
--- capacity 0, round 2 ---
  [conn 2] C->S Parse(sqlx_s_2, "select $1::int4 * 2"), Describe(S, sqlx_s_2), Sync
  [conn 2] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
  [conn 2] C->S Bind(portal unnamed, stmt sqlx_s_2), Execute(limit 0), Close(P, unnamed), Sync
  [conn 2] S->C BindComplete, DataRow(列长 4), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
capacity 0: cached = 0, server has ["sqlx_s_1", "sqlx_s_2"]

同一条 SQL 两次执行，服务端留下了 sqlx_s_1 和 sqlx_s_2，客户端缓存是空的。一条长寿的连接跑上几百万次查询，服务端这个后端进程的内存就一直涨。这正是 issue #4328（「PgConnectOptions::statement_cache_capacity(0) causes PG connection backends (in db server) to leak memory」）报告的现象，修复 PR #4337（「Fix PostgreSQL memory leak when statement cache is disabled」）在 2026-08-17 合并，晚于 0.9.0，说明里写的是「With this fix unnamed statements are used instead」。0.9.0 上想彻底不缓存，应该给每条查询写 .persistent(false)，而不是把容量设成 0。

带着元数据的 Statement：Parse 不带 Describe

prepare 里发不发 Describe，取决于调用者有没有给元数据：

    if metadata.is_none() {
        // get the statement columns and parameters
        conn.inner
            .stream
            .write_msg(message::Describe::Statement(id))?;
    }

元数据来自 Execute::statement()。Executor::prepare 返回的 PgStatement 带着一个 Arc<PgStatementMetadata>，用 stmt.query() 构造的查询把它交给 run。同一条连接上它一定命中缓存，看不出区别；拿到另一条连接上执行时，缓存没命中，就会走到这里。src/bin/unnamed.rs 的最后一段：

  // a Statement prepared on one connection, executed on another
  let stmt = conn.prepare("select $1::int4 - 1".into_sql_str()).await?;
  let mut other = PgConnection::connect_with(&opts).await?;
  quiet(&p, false);
  println!("--- Statement from conn 1 executed on a new connection ---");
  let v: i32 = stmt.query_scalar().bind(5_i32).fetch_one(&mut other).await?;
  settle().await;
  println!("result {v}");

--- Statement from conn 1 executed on a new connection ---
  [conn 3] C->S Parse(sqlx_s_1, "select $1::int4 - 1"), Sync
  [conn 3] S->C ParseComplete, ReadyForQuery(I)
  [conn 3] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute(limit 0), Close(P, unnamed), Sync
  [conn 3] S->C BindComplete, DataRow(列长 4), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
result 4

新连接上只发了 Parse 和 Sync，回来的也只有 ParseComplete 和 ReadyForQuery，参数和列的描述直接用了 conn 1 那边拿到的。语句在新连接上被命名为 sqlx_s_1（每条连接自己编号），并进了新连接的缓存。

简单查询的分支

take_arguments() 返回 None（直接把 &str 交给执行器，或 raw_sql）时，run 走另一支：

        } else {
            // Query will trigger a ReadyForQuery
            self.queue_simple_query(sql)?;

            // metadata starts out as "nothing"
            metadata = Arc::new(PgStatementMetadata::default());

            // and unprepared statements are text
            PgValueFormat::Text
        };

不查缓存，不预备，结果是文本格式。一条 Query 里可以有多条语句，每条返回行的语句，结果都以一个 RowDescription 开头，run 的读循环遇到 RowDescription 就重新解析一份元数据，替换掉 metadata，后面的 DataRow 用新的那份。这也是为什么简单查询的结果里列类型只能从 RowDescription 来，没有参数类型。什么时候走这一支、query() 不 bind 为什么不走，见 Executor 与 Acquire；多条语句中间出错时会怎样，见消息与读写缓冲的动脑筋。

fetch_optional（也就是 fetch_one）不会提前结束：它记下第一行，然后把这个 stream 一直读到 ReadyForQuery，所以只要不出错、不被取消，它不留账。代码和来历见一次查询的全过程。

动脑筋答案

--- q1 ---
  [conn 1] C->S Parse(sqlx_s_1, "select 1"), Describe(S, sqlx_s_1), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute(limit 0), Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
--- q2 ---
  [conn 1] C->S Parse(sqlx_s_2, "select 2"), Describe(S, sqlx_s_2), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_2), Execute(limit 0), Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
--- q1 again ---
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute(limit 0), Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
--- q3 ---
  [conn 1] C->S Parse(sqlx_s_3, "select 3"), Describe(S, sqlx_s_3), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
  [conn 1] C->S Close(S, sqlx_s_2), Sync
  [conn 1] S->C CloseComplete, ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_3), Execute(limit 0), Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
cached = 2, server has ["sqlx_s_1", "sqlx_s_3"]
  • 被关掉的是 sqlx_s_2。sqlx_s_1 建得更早，但第三次查询命中它时 get_mut 把它挪到了「最近使用」一端，最久没用的变成了 sqlx_s_2。如果是按插入顺序淘汰（FIFO），关掉的会是 sqlx_s_1。
  • q3 是三个来回：prepare 一个，Close(S, sqlx_s_2) 加 Sync 一个，Bind 那一组一个。Close 发生在 prepare 之后、Bind 之前：新语句先建好，确认能用，才淘汰旧的。
  • 最后 pg_prepared_statements 里正好是缓存里的两条。查它的那条语句用了 persistent(false)，走未命名语句，不会把 sqlx_s_1 挤掉。

小结

  • pending_ready_for_query_count 记着欠服务端几个 ReadyForQuery：write_sync 和 queue_simple_query 加一，收到时减一（checked_sub）。wait_until_ready 先 flush 再读到清零，run 每次开头都调它。
  • stream 读一半被 drop、future 被取消、出错、Transaction 未提交就 drop，都靠下一次 wait_until_ready 收尾。没读完的结果集仍然会被全部传过来、读进来再扔掉；收尾时遇到的 ErrorResponse 会由下一条无关的查询报出来。sqlx 不发 CancelRequest。
  • StatementCache 是按 SQL 原文的 LRU，默认 100 条。挤掉一条时单独一个来回发 Close::Statement，客户端缓存和服务端预备语句一一对应；clear_cached_statements 一个来回关掉全部。
  • persistent(false) 用未命名语句（0.8.6 起，#3863），每次都重新 Parse；但如果同一条 SQL 已经在缓存里，它照样用缓存的命名语句。
  • statement_cache_capacity(0) 在 0.9.0 里每次执行都建一个永不关闭的 sqlx_s_N（#4328，修复在 0.9.0 之后合并）；不想缓存就用 persistent(false)。
  • 拿着 PgStatement 去别的连接执行，只发 Parse，不发 Describe。

💬 关联讨论

发起讨论
加载中...