perf: sqlite prepare cached

changelog.d/sqlite-prepare-cached.changed

@@ -0,0 +1,2 @@
+Cache prepared statements on hot SQLite read paths (Clarity side-store reads, MARF block-id/hash lookups, and the generic `query_*` helpers) instead of re-parsing the SQL on every call, and raise the per-connection statement-cache capacity so the cache is not LRU-thrashed 
+

clarity/src/vm/database/sqlite.rs

@@ -54,7 +54,10 @@ pub struct SqliteConnection {
 
 fn sqlite_put(conn: &Connection, key: &str, value: &str) -> Result<(), VmExecutionError> {
     let params = params![key, value];
-    match conn.execute("REPLACE INTO data_table (key, value) VALUES (?, ?)", params) {
+    match conn
+        .prepare_cached("REPLACE INTO data_table (key, value) VALUES (?, ?)")
+        .and_then(|mut stmt| stmt.execute(params))
+    {
         Ok(_) => Ok(()),
         Err(e) => {
             error!("Failed to insert/replace ({key},{value}): {e:?}");
@@ -67,11 +70,8 @@ fn sqlite_get(conn: &Connection, key: &str) -> Result<Option<String>, VmExecutio
     trace!("sqlite_get {key}");
     let params = params![key];
     let res = match conn
-        .query_row(
-            "SELECT value FROM data_table WHERE key = ?",
-            params,
-            |row| row.get(0),
-        )
+        .prepare_cached("SELECT value FROM data_table WHERE key = ?")
+        .and_then(|mut stmt| stmt.query_row(params, |row| row.get(0)))
         .optional()
     {
         Ok(x) => Ok(x),
@@ -178,10 +178,10 @@ impl SqliteConnection {
         let key = Self::make_metadata_key(contract_id, key);
         let params = params![bhh, key, value];
 
-        if let Err(e) = conn.execute(
-            "INSERT INTO metadata_table (blockhash, key, value) VALUES (?, ?, ?)",
-            params,
-        ) {
+        if let Err(e) = conn
+            .prepare_cached("INSERT INTO metadata_table (blockhash, key, value) VALUES (?, ?, ?)")
+            .and_then(|mut stmt| stmt.execute(params))
+        {
             error!("Failed to insert ({bhh},{key},{value}): {e:?}");
             return Err(VmInternalError::DBError(SQL_FAIL_MESSAGE.into()).into());
         }
@@ -282,11 +282,8 @@ impl SqliteConnection {
         let params = params![bhh, key];
 
         match conn
-            .query_row(
-                "SELECT value FROM metadata_table WHERE blockhash = ? AND key = ?",
-                params,
-                |row| row.get(0),
-            )
+            .prepare_cached("SELECT value FROM metadata_table WHERE blockhash = ? AND key = ?")
+            .and_then(|mut stmt| stmt.query_row(params, |row| row.get(0)))
             .optional()
         {
             Ok(x) => Ok(x),
@@ -360,6 +357,10 @@ impl SqliteConnection {
         conn.busy_handler(Some(tx_busy_handler))
             .map_err(|x| VmInternalError::SqliteError(IncomparableError { err: x }))?;
 
+        // rusqlite's default statement cache holds only 16 entries; bump it so the
+        // per-read `prepare_cached` statements in this module are never LRU-evicted.
+        conn.set_prepared_statement_cache_capacity(32);
+
         Ok(conn)
     }
 }

stackslib/src/chainstate/burn/db/sortdb.rs

@@ -4905,11 +4905,12 @@ impl SortitionDB {
         // is empty -- i.e. on migration to schema 11.
         let mut cursor = tip.clone();
         for _ in 0..STACKS_TIPS_BY_BURN_VIEW_SEARCH_DEPTH {
-            let result_at_tip : Option<(ConsensusHash, ConsensusHash, BlockHeaderHash, u64)> = conn.query_row_and_then(
-                "SELECT consensus_hash,burn_view_consensus_hash, block_hash,block_height FROM stacks_chain_tips_by_burn_view WHERE sortition_id = ? ORDER BY block_height DESC LIMIT 1",
-                &[&cursor.sortition_id],
-                |row| Ok((row.get_unwrap(0), row.get_unwrap(1), row.get_unwrap(2), (u64::try_from(row.get_unwrap::<_, i64>(3)).expect("FATAL: block height too high"))))
-            ).optional()?;
+            let result_at_tip : Option<(ConsensusHash, ConsensusHash, BlockHeaderHash, u64)> = conn
+                .prepare_cached("SELECT consensus_hash,burn_view_consensus_hash, block_hash,block_height FROM stacks_chain_tips_by_burn_view WHERE sortition_id = ? ORDER BY block_height DESC LIMIT 1")?
+                .query_row(
+                    &[&cursor.sortition_id],
+                    |row| Ok((row.get_unwrap(0), row.get_unwrap(1), row.get_unwrap(2), (u64::try_from(row.get_unwrap::<_, i64>(3)).expect("FATAL: block height too high"))))
+                ).optional()?;
             test_debug!(
                 "Result at tip by burn view ({} {} {}): {:?}",
                 &cursor.sortition_id,
@@ -4931,11 +4932,12 @@ impl SortitionDB {
         // exhaustively search stacks_chain_tips
         let mut cursor = tip.clone();
         loop {
-            let result_at_tip : Option<(ConsensusHash, BlockHeaderHash, u64)> = conn.query_row_and_then(
-                "SELECT consensus_hash,block_hash,block_height FROM stacks_chain_tips WHERE sortition_id = ? ORDER BY block_height DESC LIMIT 1",
-                &[&cursor.sortition_id],
-                |row| Ok((row.get_unwrap(0), row.get_unwrap(1), (u64::try_from(row.get_unwrap::<_, i64>(2)).expect("FATAL: block height too high"))))
-            ).optional()?;
+            let result_at_tip : Option<(ConsensusHash, BlockHeaderHash, u64)> = conn
+                .prepare_cached("SELECT consensus_hash,block_hash,block_height FROM stacks_chain_tips WHERE sortition_id = ? ORDER BY block_height DESC LIMIT 1")?
+                .query_row(
+                    &[&cursor.sortition_id],
+                    |row| Ok((row.get_unwrap(0), row.get_unwrap(1), (u64::try_from(row.get_unwrap::<_, i64>(2)).expect("FATAL: block height too high"))))
+                ).optional()?;
             test_debug!(
                 "Result at tip ({} {} {}): {:?}",
                 &cursor.sortition_id,

stackslib/src/chainstate/stacks/index/trie_sql.rs

@@ -226,11 +226,8 @@ pub fn read_squashed_block_root_hash_by_height(
     height: u32,
 ) -> Result<Option<TrieHash>, Error> {
     let result: Option<Vec<u8>> = conn
-        .query_row(
-            "SELECT marf_root_hash FROM marf_squashed_blocks WHERE height = ?1",
-            params![i64::from(height)],
-            |row| row.get(0),
-        )
+        .prepare_cached("SELECT marf_root_hash FROM marf_squashed_blocks WHERE height = ?1")?
+        .query_row(params![i64::from(height)], |row| row.get(0))
         .optional()?;
 
     match result {
@@ -255,11 +252,8 @@ pub fn read_squashed_block_root_hash_by_hash<T: MarfTrieId>(
     block_hash: &T,
 ) -> Result<Option<TrieHash>, Error> {
     let result: Option<Vec<u8>> = conn
-        .query_row(
-            "SELECT marf_root_hash FROM marf_squashed_blocks WHERE block_hash = ?1",
-            params![block_hash.as_bytes()],
-            |row| row.get(0),
-        )
+        .prepare_cached("SELECT marf_root_hash FROM marf_squashed_blocks WHERE block_hash = ?1")?
+        .query_row(params![block_hash.as_bytes()], |row| row.get(0))
         .optional()?;
 
     match result {
@@ -304,11 +298,8 @@ pub fn read_squashed_block_hash_by_height<T: MarfTrieId>(
     height: u32,
 ) -> Result<Option<T>, Error> {
     let result: Option<Vec<u8>> = conn
-        .query_row(
-            "SELECT block_hash FROM marf_squashed_blocks WHERE height = ?1",
-            params![i64::from(height)],
-            |row| row.get(0),
-        )
+        .prepare_cached("SELECT block_hash FROM marf_squashed_blocks WHERE height = ?1")?
+        .query_row(params![i64::from(height)], |row| row.get(0))
         .optional()?;
 
     match result {
@@ -539,12 +530,9 @@ pub fn ensure_no_migration_necessary<T: MarfTrieId>(conn: &mut Connection) -> Re
 }
 
 pub fn get_block_identifier<T: MarfTrieId>(conn: &Connection, bhh: &T) -> Result<u32, Error> {
-    conn.query_row(
-        "SELECT block_id FROM marf_data WHERE block_hash = ?",
-        &[bhh],
-        |row| row.get("block_id"),
-    )
-    .map_err(|e| e.into())
+    conn.prepare_cached("SELECT block_id FROM marf_data WHERE block_hash = ?")?
+        .query_row(&[bhh], |row| row.get("block_id"))
+        .map_err(|e| e.into())
 }
 
 pub fn get_mined_block_identifier<T: MarfTrieId>(conn: &Connection, bhh: &T) -> Result<u32, Error> {
@@ -560,26 +548,20 @@ pub fn get_confirmed_block_identifier<T: MarfTrieId>(
     conn: &Connection,
     bhh: &T,
 ) -> Result<Option<u32>, Error> {
-    conn.query_row(
-        "SELECT block_id FROM marf_data WHERE block_hash = ? AND unconfirmed = 0",
-        &[bhh],
-        |row| row.get("block_id"),
-    )
-    .optional()
-    .map_err(|e| e.into())
+    conn.prepare_cached("SELECT block_id FROM marf_data WHERE block_hash = ? AND unconfirmed = 0")?
+        .query_row(&[bhh], |row| row.get("block_id"))
+        .optional()
+        .map_err(|e| e.into())
 }
 
 pub fn get_unconfirmed_block_identifier<T: MarfTrieId>(
     conn: &Connection,
     bhh: &T,
 ) -> Result<Option<u32>, Error> {
-    conn.query_row(
-        "SELECT block_id FROM marf_data WHERE block_hash = ? AND unconfirmed = 1",
-        &[bhh],
-        |row| row.get("block_id"),
-    )
-    .optional()
-    .map_err(|e| e.into())
+    conn.prepare_cached("SELECT block_id FROM marf_data WHERE block_hash = ? AND unconfirmed = 1")?
+        .query_row(&[bhh], |row| row.get("block_id"))
+        .optional()
+        .map_err(|e| e.into())
 }
 
 pub fn get_latest_confirmed_block_hash<T: MarfTrieId>(conn: &Connection) -> Result<T, Error> {
@@ -593,11 +575,8 @@ pub fn get_latest_confirmed_block_hash<T: MarfTrieId>(conn: &Connection) -> Resu
 
 pub fn get_block_hash<T: MarfTrieId>(conn: &Connection, local_id: u32) -> Result<T, Error> {
     let result = conn
-        .query_row(
-            "SELECT block_hash FROM marf_data WHERE block_id = ?",
-            params![local_id],
-            |row| row.get("block_hash"),
-        )
+        .prepare_cached("SELECT block_hash FROM marf_data WHERE block_id = ?")?
+        .query_row(params![local_id], |row| row.get("block_hash"))
         .optional()?;
     result.ok_or_else(|| {
         error!("Failed to get block header hash of local ID {}", local_id);
@@ -1025,11 +1004,11 @@ pub fn lock_bhh_for_extension<T: MarfTrieId>(
 }
 
 pub fn count_blocks(conn: &Connection) -> Result<u32, Error> {
-    let result = conn.query_row(
-        "SELECT IFNULL(MAX(block_id), 0) AS count FROM marf_data WHERE unconfirmed = 0",
-        NO_PARAMS,
-        |row| row.get("count"),
-    )?;
+    let result = conn
+        .prepare_cached(
+            "SELECT IFNULL(MAX(block_id), 0) AS count FROM marf_data WHERE unconfirmed = 0",
+        )?
+        .query_row(NO_PARAMS, |row| row.get("count"))?;
     Ok(result)
 }
 

stackslib/src/clarity_vm/database/mod.rs

@@ -697,11 +697,13 @@ where
         "block_headers"
     };
 
-    conn.query_row(
-        &format!("SELECT {column_name} FROM {table_name} WHERE index_block_hash = ?",),
-        args,
-        |x| Ok(loader(x)),
-    )
+    // `column_name`/`table_name` come from small fixed sets (7 as of this
+    // writing) times the two header tables, far below the connection's
+    // 200-entry statement-cache capacity.
+    conn.prepare_cached(&format!(
+        "SELECT {column_name} FROM {table_name} WHERE index_block_hash = ?"
+    ))
+    .and_then(|mut stmt| stmt.query_row(args, |x| Ok(loader(x))))
     .optional()
     .unwrap_or_else(|_| {
         panic!(
@@ -822,14 +824,15 @@ fn get_matured_reward<GTS: GetTenureStartId>(
     };
     let parent_id_bhh = conn
         .conn()
-        .query_row(
-            &format!("SELECT parent_block_id FROM {table_name} WHERE index_block_hash = ?"),
-            params![child_id_bhh.0],
-            |x| {
+        .prepare_cached(&format!(
+            "SELECT parent_block_id FROM {table_name} WHERE index_block_hash = ?"
+        ))
+        .and_then(|mut stmt| {
+            stmt.query_row(params![child_id_bhh.0], |x| {
                 Ok(StacksBlockId::from_column(x, "parent_block_id")
                     .expect("Bad parent_block_id in database"))
-            },
-        )
+            })
+        })
         .optional()
         .expect("Unexpected SQL failure querying parent block ID");
 

stackslib/src/util_lib/db.rs

@@ -398,7 +398,7 @@ where
     T: FromRow<T>,
 {
     log_sql_eqp(conn, sql_query);
-    let mut stmt = conn.prepare(sql_query)?;
+    let mut stmt = conn.prepare_cached(sql_query)?;
     let result = stmt.query_and_then(sql_args, |row| T::from_row(row))?;
 
     result.collect()
@@ -412,11 +412,11 @@ where
     T: FromRow<T>,
 {
     log_sql_eqp(conn, sql_query);
-    let query_result = conn.query_row_and_then(sql_query, sql_args, |row| T::from_row(row));
-    match query_result {
-        Ok(x) => Ok(Some(x)),
-        Err(Error::SqliteError(sqlite_error::QueryReturnedNoRows)) => Ok(None),
-        Err(e) => Err(e),
+    let mut stmt = conn.prepare_cached(sql_query)?;
+    let mut result = stmt.query_and_then(sql_args, |row| T::from_row(row))?;
+    match result.next() {
+        Some(value) => Ok(Some(value?)),
+        None => Ok(None),
     }
 }
 
@@ -432,7 +432,7 @@ where
     T: FromRow<T>,
 {
     log_sql_eqp(conn, sql_query);
-    let mut stmt = conn.prepare(sql_query)?;
+    let mut stmt = conn.prepare_cached(sql_query)?;
     let mut result = stmt.query_and_then(sql_args, |row| T::from_row(row))?;
     let mut return_value = None;
     if let Some(value) = result.next() {
@@ -458,7 +458,7 @@ where
     F: FnOnce() -> String,
 {
     log_sql_eqp(conn, sql_query);
-    let mut stmt = conn.prepare(sql_query)?;
+    let mut stmt = conn.prepare_cached(sql_query)?;
     let mut result = stmt.query_and_then(sql_args, |row| T::from_row(row))?;
     let mut return_value = None;
     if let Some(value) = result.next() {
@@ -482,7 +482,7 @@ where
     T: FromColumn<T>,
 {
     log_sql_eqp(conn, sql_query);
-    let mut stmt = conn.prepare(sql_query)?;
+    let mut stmt = conn.prepare_cached(sql_query)?;
     let mut rows = stmt.query(sql_args)?;
 
     // gather
@@ -509,7 +509,7 @@ where
     T: FromColumn<T>,
 {
     log_sql_eqp(conn, sql_query);
-    let mut stmt = conn.prepare(sql_query)?;
+    let mut stmt = conn.prepare_cached(sql_query)?;
     let mut rows = stmt.query(sql_args)?;
 
     // gather
@@ -531,7 +531,7 @@ where
     P: Params,
 {
     log_sql_eqp(conn, sql_query);
-    let mut stmt = conn.prepare(sql_query)?;
+    let mut stmt = conn.prepare_cached(sql_query)?;
     let mut rows = stmt.query(sql_args)?;
     let mut row_data = None;
     while let Some(row) = rows.next().map_err(Error::SqliteError)? {
@@ -746,6 +746,10 @@ pub fn sqlite_open<P: AsRef<Path>>(
     flags.insert(OpenFlags::SQLITE_OPEN_NO_MUTEX);
     let db = inner_connection_open(path, flags)?;
     db.busy_handler(Some(tx_busy_handler))?;
+    // The node's long-lived connections run far more distinct statements than
+    // rusqlite's default cache of 16 holds (sortition, headers, staging, MARF, ...),
+    // so `prepare_cached` would LRU-thrash and re-parse anyway.
+    db.set_prepared_statement_cache_capacity(200);
     if !flags.contains(OpenFlags::SQLITE_OPEN_READ_ONLY) {
         inner_sql_pragma(&db, "journal_mode", &"WAL")?;
     }
@@ -786,7 +790,7 @@ pub fn get_ancestor_block_height<T: MarfTrieId>(
 /// Load some index data
 fn load_indexed(conn: &DBConn, marf_value: &MARFValue) -> Result<Option<String>, Error> {
     let mut stmt = conn
-        .prepare("SELECT value FROM __fork_storage WHERE value_hash = ?1 LIMIT 2")
+        .prepare_cached("SELECT value FROM __fork_storage WHERE value_hash = ?1 LIMIT 2")
         .map_err(Error::SqliteError)?;
     let mut rows = stmt
         .query(params![marf_value.to_hex()])