fix: ensure oasysdb works well with postgres (#109)

Cargo.toml

@@ -18,6 +18,7 @@ categories = ["database", "algorithms", "embedded"]
 [dependencies]
 uuid = { version = "1.9.1", features = ["v4", "fast-rng", "serde"] }
 half = { version = "2.4.1", features = ["serde"] }
+tokio = { version = "1.39.2", features = ["rt-multi-thread"] }
 url = "2.5.2"
 futures = "0.3.30"
 rand = "0.8.5"
@@ -34,7 +35,7 @@ serde_json = "1.0.120"
 [dependencies.sqlx]
 version = "0.7.4"
 default-features = false
-features = ["all-databases"]
+features = ["all-databases", "runtime-tokio"]
 
 [dev-dependencies]
 byteorder = "1.5.0"

examples/measure_recall.rs

@@ -1,7 +1,7 @@
 use common::Dataset;
-use futures::executor;
 use oasysdb::prelude::*;
 use std::error::Error;
+use tokio::runtime::Runtime;
 
 mod common;
 
@@ -10,7 +10,8 @@ fn main() -> Result<(), Box<dyn Error>> {
     let db_url = dataset.database_url();
     let config = SourceConfig::new(dataset.name(), "id", "vector");
 
-    executor::block_on(dataset.populate_database())?;
+    let rt = Runtime::new()?;
+    rt.block_on(dataset.populate_database())?;
 
     let db = Database::open("odb_example", Some(db_url))?;
     create_index_flat(&db, &config)?;

src/db/database.rs

@@ -1,9 +1,9 @@
 use super::*;
-use futures::executor;
 use futures::stream::StreamExt;
 use sqlx::any::install_default_drivers;
 use sqlx::Acquire;
 use std::sync::{Arc, Mutex};
+use tokio::runtime::Runtime;
 use url::Url;
 use uuid::Uuid;
 
@@ -159,7 +159,8 @@ impl Database {
         algorithm: IndexAlgorithm,
         config: SourceConfig,
     ) -> Result<(), Error> {
-        executor::block_on(self.async_create_index(name, algorithm, config))
+        let rt = Runtime::new()?;
+        rt.block_on(self.async_create_index(name, algorithm, config))
     }
 
     /// Returns an index reference.
@@ -263,7 +264,8 @@ impl Database {
     /// Updates the index with new records from the source synchronously.
     /// - `name`: Index name.
     pub fn refresh_index(&self, name: impl AsRef<str>) -> Result<(), Error> {
-        executor::block_on(self.async_refresh_index(name))
+        let rt = Runtime::new()?;
+        rt.block_on(self.async_refresh_index(name))
     }
 
     /// Searches the index for nearest neighbors.
@@ -417,7 +419,8 @@ impl DatabaseState {
 
     /// Connects to the source SQL database.
     pub fn connect(&self) -> Result<SourceConnection, Error> {
-        executor::block_on(self.async_connect())
+        let rt = Runtime::new()?;
+        rt.block_on(self.async_connect())
     }
 
     /// Disconnects from the source SQL database asynchronously.
@@ -429,7 +432,8 @@ impl DatabaseState {
     /// Disconnects from the source SQL database.
     /// - `conn`: Database connection.
     pub fn disconnect(conn: SourceConnection) -> Result<(), Error> {
-        executor::block_on(Self::async_disconnect(conn))
+        let rt = Runtime::new()?;
+        rt.block_on(Self::async_disconnect(conn))
     }
 
     /// Validates the connection to the source database.
@@ -539,7 +543,9 @@ mod tests {
     fn test_database_refresh_index() -> Result<(), Error> {
         let db = create_test_database()?;
         let query = generate_insert_query(100, 10);
-        executor::block_on(db.async_execute_sql(query))?;
+
+        let rt = Runtime::new()?;
+        rt.block_on(db.async_execute_sql(query))?;
 
         db.refresh_index(TEST_INDEX).unwrap();
 
@@ -616,7 +622,9 @@ mod tests {
         let state = db.state()?;
         assert_eq!(state.source_type(), SourceType::SQLITE);
 
-        executor::block_on(setup_test_source(&db_url))?;
+        let rt = Runtime::new()?;
+        rt.block_on(setup_test_source(&db_url))?;
+
         create_test_index(&mut db)?;
         Ok(db)
     }