Add LIMIT clause to fulltext scans to prevent OOM issues (#202)

Author: emmanuel-keller

.github/workflows/ci.yml

@@ -24,7 +24,6 @@ jobs:
   format:
     name: Check format
     runs-on: [runner-amd64-large]
-    continue-on-error: true
     steps:
       - name: Install stable toolchain
         uses: dtolnay/rust-toolchain@stable
@@ -46,7 +45,6 @@ jobs:
   clippy:
     name: Check clippy
     runs-on: [runner-amd64-large]
-    continue-on-error: true
     steps:
       - name: Install stable toolchain
         uses: dtolnay/rust-toolchain@stable
@@ -68,7 +66,6 @@ jobs:
   test:
     name: Check tests
     runs-on: [runner-amd64-large]
-    continue-on-error: true
     steps:
       - name: Install stable toolchain
         uses: dtolnay/rust-toolchain@stable
@@ -97,7 +94,6 @@ jobs:
             runner: runner-arm64-large
             target: aarch64-unknown-linux-gnu
     runs-on: ${{ matrix.runner }}
-    continue-on-error: true
     steps:
       - name: Install stable toolchain
         uses: dtolnay/rust-toolchain@stable
@@ -124,7 +120,6 @@ jobs:
     name: Build Docker image
     runs-on: [runner-amd64-large-private]
     needs: build
-    continue-on-error: true
     steps:
       - name: Checkout sources
         uses: actions/checkout@v4
@@ -194,7 +189,6 @@ jobs:
     name: Benchmark ${{ matrix.description }}
     needs: build
     runs-on: [ubuntu-latest]
-    continue-on-error: true
     env:
       CRUD_BENCH_BATCHES: >-
         [

src/main.rs

@@ -181,7 +181,7 @@ pub(crate) struct Args {
 					"fields": ["number"]
 				}
 			},
-			{ "name": "where_field_fulltext_single", "samples": 100, "projection": "FULL",
+			{ "name": "where_field_fulltext_single", "samples": 100, "projection": "FULL", "limit": 1000,
 				"condition": {
 					"sql": "to_tsvector('english', words) @@ to_tsquery('english', 'hello')",
 					"mysql": "MATCH(words) AGAINST('hello' IN NATURAL LANGUAGE MODE)",
@@ -194,7 +194,7 @@ pub(crate) struct Args {
 					"index_type": "fulltext"
 				}
 			},
-			{ "name": "where_field_fulltext_multi_and", "samples": 100, "projection": "FULL",
+			{ "name": "where_field_fulltext_multi_and", "samples": 100, "projection": "FULL", "limit": 1000,
 				"condition": {
 					"sql": "to_tsvector('english', words) @@ to_tsquery('english', 'hello & world')",
 					"mysql": "MATCH(words) AGAINST('+hello +world' IN NATURAL LANGUAGE MODE)",
@@ -207,7 +207,7 @@ pub(crate) struct Args {
 					"index_type": "fulltext"
 				}
 			},
-			{ "name": "where_field_fulltext_multi_or", "samples": 100, "projection": "FULL",
+			{ "name": "where_field_fulltext_multi_or", "samples": 100, "projection": "FULL", "limit": 1000,
 				"condition": {
 					"sql": "to_tsvector('english', words) @@ to_tsquery('english', 'foo | bar')",
 					"mysql": "MATCH(words) AGAINST('foo bar' IN NATURAL LANGUAGE MODE)",

src/neo4j.rs

@@ -238,7 +238,7 @@ impl Neo4jClient {
 			Projection::Id => {
 				let stm = match fts {
 					true => format!(
-						"CALL db.index.fulltext.queryNodes('{n}', '{c}') YIELD node as r {s} {l} RETURN r.id"
+						"CALL db.index.fulltext.queryNodes('{n}', '{c}') YIELD node as r WITH r {s} {l} RETURN r.id"
 					),
 					false => format!("MATCH (r) {c} {s} {l} RETURN r.id"),
 				};
@@ -253,7 +253,7 @@ impl Neo4jClient {
 			Projection::Full => {
 				let stm = match fts {
 					true => format!(
-						"CALL db.index.fulltext.queryNodes('{n}', '{c}') YIELD node as r {s} {l} RETURN r"
+						"CALL db.index.fulltext.queryNodes('{n}', '{c}') YIELD node as r WITH r {s} {l} RETURN r"
 					),
 					false => format!("MATCH (r) {c} {s} {l} RETURN r"),
 				};
@@ -268,7 +268,7 @@ impl Neo4jClient {
 			Projection::Count => {
 				let stm = match fts {
 					true => format!(
-						"CALL db.index.fulltext.queryNodes('{n}', '{c}') YIELD node as r {s} {l} RETURN count(r) as count"
+						"CALL db.index.fulltext.queryNodes('{n}', '{c}') YIELD node as r WITH r {s} {l} RETURN count(r) as count"
 					),
 					false => format!("MATCH (r) {c} {s} {l} RETURN count(r) as count"),
 				};

src/sqlite.rs

@@ -13,9 +13,9 @@ use std::cmp::max;
 use std::hint::black_box;
 use std::sync::Arc;
 use std::time::Duration;
-use tokio_rusqlite::Connection;
 use tokio_rusqlite::types::ToSqlOutput;
 use tokio_rusqlite::types::Value;
+use tokio_rusqlite::{Connection, rusqlite};
 
 const DATABASE_DIR: &str = "sqlite";
 
@@ -265,7 +265,7 @@ impl BenchmarkClient for SqliteClient {
 
 impl SqliteClient {
 	async fn execute_batch(&self, query: Cow<'static, str>) -> Result<()> {
-		self.conn.call(move |conn| conn.execute_batch(query.as_ref()).map_err(Into::into)).await?;
+		self.conn.call(move |conn| conn.execute_batch(query.as_ref())).await?;
 		Ok(())
 	}
 
@@ -275,7 +275,7 @@ impl SqliteClient {
 		params: ToSqlOutput<'static>,
 	) -> Result<usize> {
 		self.conn
-			.call(move |conn| conn.execute(query.as_ref(), [&params]).map_err(Into::into))
+			.call(move |conn| conn.execute(query.as_ref(), [&params]))
 			.await
 			.map_err(Into::into)
 	}
@@ -286,7 +286,7 @@ impl SqliteClient {
 		params: Option<ToSqlOutput<'static>>,
 	) -> Result<Vec<Row>> {
 		self.conn
-			.call(move |conn| {
+			.call(move |conn| -> rusqlite::Result<Vec<Row>> {
 				let mut stmt = conn.prepare(stmt.as_ref())?;
 				let mut rows = match params {
 					Some(params) => stmt.query([&params])?,
@@ -424,7 +424,7 @@ impl SqliteClient {
 		let conn = self.conn.clone();
 		let column_defs = self.columns.clone();
 
-		conn.call(move |conn| {
+		conn.call(move |conn| -> rusqlite::Result<()> {
 			// Store the records to insert
 			let mut inserts = Vec::with_capacity(key_vals.len());
 			// Store all parameter values as owned types
@@ -482,7 +482,7 @@ impl SqliteClient {
 	{
 		let conn = self.conn.clone();
 
-		conn.call(move |conn| {
+		conn.call(move |conn| -> rusqlite::Result<()> {
 			// Build the IN clause with positional parameters
 			let ids = (1..=keys.len()).map(|i| format!("${i}")).collect::<Vec<String>>().join(", ");
 
@@ -540,7 +540,7 @@ impl SqliteClient {
 		let conn = self.conn.clone();
 		let column_defs = self.columns.clone();
 
-		conn.call(move |conn| {
+		conn.call(move |conn| -> rusqlite::Result<()> {
 			// For SQLite, we'll use multiple UPDATE statements in a transaction
 			// since SQLite doesn't support UPDATE FROM as elegantly as PostgreSQL
 
@@ -609,7 +609,7 @@ impl SqliteClient {
 	{
 		let conn = self.conn.clone();
 
-		conn.call(move |conn| {
+		conn.call(move |conn| -> rusqlite::Result<()> {
 			// Build the IN clause with positional parameters
 			let ids = (1..=keys.len()).map(|i| format!("${i}")).collect::<Vec<String>>().join(", ");
 

src/surrealdb.rs

@@ -9,6 +9,7 @@ use crate::memory::Config as MemoryConfig;
 use crate::valueprovider::Columns;
 use crate::{Benchmark, Index, KeyType, Projection, Scan};
 use anyhow::{Result, bail};
+use log::warn;
 use serde_json::Value;
 use std::time::Duration;
 use surrealdb::Surreal;
@@ -258,12 +259,68 @@ impl BenchmarkClient for SurrealDBClient {
 	}
 
 	async fn drop_index(&self, name: &str) -> Result<()> {
+		// Retry helper closure for handling transient "Resource busy" errors.
+		//
+		// ## Why Retry is Necessary
+		//
+		// After intensive concurrent scan operations (e.g., 12 clients × 48 threads = 576 tasks),
+		// each scan creates a READ transaction in SurrealDB that holds a RocksDB snapshot.
+		// These snapshots capture the database state and provide MVCC (Multi-Version Concurrency Control).
+		//
+		// When REMOVE INDEX executes immediately after scans complete:
+		// 1. Client-side: All Rust futures have finished (via try_join_all)
+		// 2. Server-side: SurrealDB/RocksDB may still have:
+		//    - Active transaction objects not yet fully released
+		//    - Snapshot references held in memory
+		//    - Deferred cleanup operations in progress
+		//
+		// ## The Conflict
+		//
+		// REMOVE INDEX runs in a WRITE transaction that needs to:
+		// - Delete index metadata keys (del_tb_index)
+		// - Update table definition (put_tb)
+		// - Clear caches
+		//
+		// RocksDB's optimistic transaction engine detects conflicts between:
+		// - Active READ snapshots from completed scan operations
+		// - WRITE transaction from REMOVE INDEX trying to modify metadata
+		//
+		// This results in: "The query was not executed due to a failed transaction. Resource busy:"
+		//
+		// ## Why Retry Works
+		//
+		// The error is transient. As transaction objects are dropped and snapshots released,
+		// the metadata locks become available. The 500ms sleep allows sufficient time for:
+		// - Async transaction cleanup to complete
+		// - RocksDB to release internal snapshot references
+		// - Memory management to finalize deferred operations
+		//
+		// Since REMOVE INDEX IF EXISTS is idempotent, retrying is safe and appropriate
+		// for benchmark scenarios where the goal is reliable completion rather than
+		// immediate failure on transient resource contention.
+		let retry = |sql: String| async move {
+			loop {
+				match self.db.query(&sql).await?.check() {
+					Ok(_) => return Ok(()),
+					Err(e) => {
+						if e.to_string().eq(
+							"The query was not executed due to a failed transaction. Resource busy: ",
+						) {
+							warn!("Retrying {sql} due to {e}");
+							sleep(Duration::from_millis(500)).await;
+						} else {
+							return Err(e);
+						}
+					}
+				}
+			}
+		};
 		// Remove the index
 		let sql = format!("REMOVE INDEX IF EXISTS {name} ON TABLE record");
-		self.db.query(sql).await?.check()?;
+		retry(sql).await?;
 		// Remove the analyzer
 		let sql = format!("REMOVE ANALYZER IF EXISTS {name}");
-		self.db.query(sql).await?.check()?;
+		retry(sql).await?;
 		// All ok
 		Ok(())
 	}