feat: add search for the runtime's /v1/search endpoint

README.md

@@ -210,6 +210,41 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
 
 To cancel an *async query job* instead, use `cancel_query()` — see [Async query jobs](#async-query-jobs-and-dataset-refresh) above.
 
+### Search
+
+`search` finds documents similar to a piece of text using the runtime's `/v1/search` endpoint. It runs against datasets that have an embedding column and a loaded embedding model — see [Search & Retrieval](https://docs.spice.ai/features/search-and-retrieval) for how to configure them. Like dataset refresh, it uses the HTTP API, so `http_url()` must be configured.
+
+```rust,no_run
+use spiceai::{ClientBuilder, SearchRequest};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+  let client = ClientBuilder::new()
+    .http_url("http://localhost:8090")
+    .build()
+    .await?;
+
+  let response = client
+    .search(
+      SearchRequest::new("tokyo plane tickets")
+        .with_datasets(["app_messages"])
+        .with_limit(3)
+        .with_additional_columns(["timestamp"]),
+    )
+    .await?;
+
+  println!("{} matches in {}ms", response.len(), response.duration_ms);
+  for m in response {
+    println!("{} {} {:?}", m.score, m.dataset, m.matches);
+  }
+  Ok(())
+}
+```
+
+Adding `with_keywords([...])` runs a lexical pass alongside the vector pass, which the runtime combines into a single hybrid ranking. `with_where("user_id = 42")` filters candidate rows with a SQL predicate.
+
+Each `SearchMatch` carries `dataset`, `score` (higher is more similar), `matches` (matched values keyed by source column — a list per column, since one column can contribute several chunks to a match), `primary_key`, `data`, and `metadata`.
+
 ## Documentation
 
 Check out our [Documentation](https://docs.spice.ai/sdks/rust-sdk) to learn more about how to use the Rust SDK.

src/client.rs

@@ -7,6 +7,7 @@ use crate::{
     config::{GenericError, SPICE_CLOUD_FLIGHT_ADDR, SPICE_LOCAL_FLIGHT_ADDR},
     dataset::{DatasetError, DatasetRefreshRequest, DatasetRefreshResponse},
     flight::{SqlFlightClient, is_connection_reset_generic_error},
+    search::{SearchError, SearchRequest, SearchResponse},
     tls::{FlightChannelBuilder, ensure_crypto_provider, new_tls_flight_channel},
 };
 use arrow::record_batch::RecordBatch;
@@ -677,6 +678,55 @@ impl SpiceClient {
 
         http_client.refresh_dataset(dataset_name, &request).await
     }
+
+    /// Searches datasets for documents similar to the request's text.
+    ///
+    /// Runs against datasets that have an embedding column and a loaded
+    /// embedding model — see [Search & Retrieval](https://docs.spice.ai/features/search-and-retrieval)
+    /// for how to configure them. Adding keywords to the request turns this
+    /// into a hybrid search, combining a lexical pass with the vector scores.
+    ///
+    /// **Note:** Requires [`http_url()`](SpiceClientBuilder::http_url) to be configured.
+    ///
+    /// # Example
+    ///
+    /// ```no_run
+    /// # use spiceai::{ClientBuilder, SearchRequest};
+    /// # #[tokio::main]
+    /// # async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    /// let client = ClientBuilder::new()
+    ///     .http_url("http://localhost:8090")
+    ///     .build()
+    ///     .await?;
+    ///
+    /// let response = client
+    ///     .search(
+    ///         SearchRequest::new("tokyo plane tickets")
+    ///             .with_datasets(["app_messages"])
+    ///             .with_limit(3),
+    ///     )
+    ///     .await?;
+    ///
+    /// for m in response {
+    ///     println!("{} {} {:?}", m.score, m.dataset, m.matches);
+    /// }
+    /// # Ok(())
+    /// # }
+    /// ```
+    ///
+    /// # Errors
+    ///
+    /// - [`SearchError::InvalidRequest`] if the request has empty text or a zero limit
+    /// - [`SearchError::HttpError`] if the HTTP endpoint is not configured or unreachable
+    /// - [`SearchError::SearchFailed`] if the runtime rejects the search
+    pub async fn search(&self, request: SearchRequest) -> Result<SearchResponse, SearchError> {
+        let http_client = self.http_client.as_ref().ok_or(SearchError::HttpError {
+            message: "HTTP endpoint not configured. Use ClientBuilder::http_url() to set it."
+                .to_string(),
+        })?;
+
+        http_client.search(&request).await
+    }
 }
 
 /// Builder for creating a `SpiceClient`.
@@ -1510,6 +1560,118 @@ mod tests {
         assert_eq!(custom_refresh.message, "Refresh scheduled");
     }
 
+    #[tokio::test]
+    async fn test_search_posts_request_and_parses_response() {
+        let server = MockServer::start().await;
+
+        Mock::given(method("POST"))
+            .and(path("/v1/search"))
+            .and(body_json(json!({
+                "text": "tokyo plane tickets",
+                "datasets": ["app_messages"],
+                "limit": 3,
+                "additional_columns": ["timestamp"],
+            })))
+            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+                "results": [
+                    {
+                        "matches": {"message": ["I booked us some tickets", "direct to Narita"]},
+                        "dataset": "app_messages",
+                        "primary_key": {"id": "6fd5a215"},
+                        "data": {"timestamp": 1_724_716_542_i64},
+                        "_score": 0.914_321
+                    },
+                    {
+                        "matches": {"message": ["we're sitting together"]},
+                        "dataset": "app_messages",
+                        "_score": 0.787_654
+                    }
+                ],
+                "duration_ms": 42
+            })))
+            .mount(&server)
+            .await;
+
+        let client = test_client(Some(&server.uri()));
+
+        let response = client
+            .search(
+                SearchRequest::new("tokyo plane tickets")
+                    .with_datasets(["app_messages"])
+                    .with_limit(3)
+                    .with_additional_columns(["timestamp"]),
+            )
+            .await
+            .expect("search succeeds");
+
+        assert_eq!(response.duration_ms, 42);
+        assert_eq!(response.len(), 2);
+
+        let first = &response.results[0];
+        assert_eq!(first.dataset, "app_messages");
+        // One column can contribute several chunks to a single match.
+        assert_eq!(first.matches["message"].len(), 2);
+        assert_eq!(first.data["timestamp"], 1_724_716_542_i64);
+
+        // The runtime omits data and primary_key when they are empty.
+        assert!(response.results[1].data.is_empty());
+        assert!(response.results[1].primary_key.is_empty());
+    }
+
+    #[tokio::test]
+    async fn test_search_surfaces_runtime_error_body() {
+        let server = MockServer::start().await;
+
+        Mock::given(method("POST"))
+            .and(path("/v1/search"))
+            .respond_with(ResponseTemplate::new(400).set_body_string("No data sources provided"))
+            .mount(&server)
+            .await;
+
+        let client = test_client(Some(&server.uri()));
+
+        let err = client
+            .search(SearchRequest::new("tokyo"))
+            .await
+            .expect_err("runtime rejects the search");
+
+        let message = err.to_string();
+        assert!(message.contains("No data sources provided"), "{message}");
+        assert!(message.contains("400"), "{message}");
+    }
+
+    #[tokio::test]
+    async fn test_search_validates_before_sending() {
+        let server = MockServer::start().await;
+        // No mock is mounted: a request reaching the server fails the test.
+        let client = test_client(Some(&server.uri()));
+
+        let err = client
+            .search(SearchRequest::new(""))
+            .await
+            .expect_err("empty text is rejected");
+        assert!(err.to_string().contains("non-empty"), "{err}");
+
+        assert!(
+            server
+                .received_requests()
+                .await
+                .unwrap_or_default()
+                .is_empty()
+        );
+    }
+
+    #[tokio::test]
+    async fn test_search_requires_http_url() {
+        let client = test_client(None);
+
+        let err = client
+            .search(SearchRequest::new("tokyo"))
+            .await
+            .expect_err("search without an HTTP endpoint fails");
+        assert!(err.to_string().contains("http_url"), "{err}");
+    }
+
     #[tokio::test]
     async fn test_active_queries_lists_from_the_runtime() {
         let server = MockServer::start().await;

src/lib.rs

@@ -8,6 +8,7 @@ mod flight;
 mod params;
 pub mod query;
 mod redirect;
+pub mod search;
 pub mod tls;
 mod util;
 
@@ -24,6 +25,7 @@ pub use query::{
     QueryError, QueryInfo, QueryJob, QueryListResponse, QueryResult, QueryResultStream,
     QueryStatus, QuerySubmitOptions, QuerySummary,
 };
+pub use search::{SearchError, SearchMatch, SearchRequest, SearchResponse};
 
 // Further public exports and integrations
 pub use futures::StreamExt;

src/query.rs

@@ -36,6 +36,7 @@
 use crate::active_query::{ActiveQueryError, ActiveQueryList, CancelActiveQueryResponse};
 use crate::dataset::{DatasetError, DatasetRefreshRequest, DatasetRefreshResponse};
 use crate::params::{QueryParameterError, QueryParameters};
+use crate::search::{SearchError, SearchRequest, SearchResponse};
 use arrow::array::RecordBatch;
 use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
 use futures::Stream;
@@ -910,6 +911,43 @@ impl QueryHttpClient {
         }
     }
 
+    pub async fn search(&self, request: &SearchRequest) -> Result<SearchResponse, SearchError> {
+        request.validate()?;
+
+        let url = format!("{}/v1/search", self.base_url);
+
+        let response = self
+            .add_auth(self.client.post(&url))
+            .json(request)
+            .send()
+            .await
+            .map_err(|e| SearchError::HttpError {
+                message: e.to_string(),
+            })?;
+
+        let status_code = response.status().as_u16();
+        if status_code != 200 {
+            // The runtime explains search failures in a plain-text body ("No
+            // data sources provided"). Surface it, not just the status code.
+            let response_body = match response.text().await {
+                Ok(body) => body.trim().to_string(),
+                // The status code is already known, so a body that cannot be read
+                // reports why instead of collapsing to an empty string — otherwise
+                // the transport failure is lost and the error reads as if the
+                // runtime had explained nothing.
+                Err(e) => format!("<error body could not be read: {e}>"),
+            };
+            return Err(SearchError::SearchFailed {
+                status_code,
+                response_body,
+            });
+        }
+
+        response.json().await.map_err(|e| SearchError::ParseError {
+            message: e.to_string(),
+        })
+    }
+
     /// List queries with optional status filter and limit.
     pub async fn list_queries(
         &self,