diff --git a/client b/client index efe9dfe0..ede3c38a 160000 --- a/client +++ b/client @@ -1 +1 @@ -Subproject commit efe9dfe05b12b2fa414c85bbaa3bd38e4d4d2078 +Subproject commit ede3c38af10a7b4ee927c00ee0ebbb5e776cbd6d diff --git a/common/proto/search.proto b/common/proto/search.proto index a5cd0335..2418b78a 100644 --- a/common/proto/search.proto +++ b/common/proto/search.proto @@ -12,6 +12,8 @@ service Searcher { // Returns live authorized row data and one-based filtered navigation bounds. // A missing position returns NOT_FOUND. Search and row reads are not a snapshot. rpc GetFilteredRow(SearchRequest) returns (komp_ac.tables_data.GetTableDataResponse); + // Up to 64 independent counts. Results preserve request order; errors are per item. + rpc BatchCount(BatchCountRequest) returns (BatchCountResponse); rpc Count(SearchRequest) returns (SearchCountResponse); } @@ -127,3 +129,18 @@ message SearchResponse { } repeated Hit hits = 1; } + +message BatchCountRequest { + repeated SearchRequest requests = 1; +} + +message BatchCountResult { + oneof outcome { + uint64 count = 1; + string error = 2; + } +} + +message BatchCountResponse { + repeated BatchCountResult results = 1; +} diff --git a/common/src/proto/descriptor.bin b/common/src/proto/descriptor.bin index c7a47173..60f31304 100644 Binary files a/common/src/proto/descriptor.bin and b/common/src/proto/descriptor.bin differ diff --git a/common/src/proto/komp_ac.search.rs b/common/src/proto/komp_ac.search.rs index e08b008f..8d8e8cae 100644 --- a/common/src/proto/komp_ac.search.rs +++ b/common/src/proto/komp_ac.search.rs @@ -150,6 +150,35 @@ pub mod search_response { } } #[derive(serde::Serialize, serde::Deserialize)] +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct BatchCountRequest { + #[prost(message, repeated, tag = "1")] + pub requests: ::prost::alloc::vec::Vec, +} +#[derive(serde::Serialize, serde::Deserialize)] +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct BatchCountResult { + #[prost(oneof = "batch_count_result::Outcome", tags = "1, 2")] + pub outcome: ::core::option::Option, +} +/// Nested message and enum types in `BatchCountResult`. +pub mod batch_count_result { + #[derive(serde::Serialize, serde::Deserialize)] + #[derive(Clone, PartialEq, Eq, Hash, ::prost::Oneof)] + pub enum Outcome { + #[prost(uint64, tag = "1")] + Count(u64), + #[prost(string, tag = "2")] + Error(::prost::alloc::string::String), + } +} +#[derive(serde::Serialize, serde::Deserialize)] +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct BatchCountResponse { + #[prost(message, repeated, tag = "1")] + pub results: ::prost::alloc::vec::Vec, +} +#[derive(serde::Serialize, serde::Deserialize)] #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] #[repr(i32)] pub enum MatchMode { @@ -379,6 +408,31 @@ pub mod searcher_client { .insert(GrpcMethod::new("komp_ac.search.Searcher", "GetFilteredRow")); self.inner.unary(req, path, codec).await } + /// Up to 64 independent counts. Results preserve request order; errors are per item. + pub async fn batch_count( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { + self.inner + .ready() + .await + .map_err(|e| { + tonic::Status::unknown( + format!("Service was not ready: {}", e.into()), + ) + })?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static( + "/komp_ac.search.Searcher/BatchCount", + ); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("komp_ac.search.Searcher", "BatchCount")); + self.inner.unary(req, path, codec).await + } pub async fn count( &mut self, request: impl tonic::IntoRequest, @@ -433,6 +487,14 @@ pub mod searcher_server { tonic::Response, tonic::Status, >; + /// Up to 64 independent counts. Results preserve request order; errors are per item. + async fn batch_count( + &self, + request: tonic::Request, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn count( &self, request: tonic::Request, @@ -603,6 +665,51 @@ pub mod searcher_server { }; Box::pin(fut) } + "/komp_ac.search.Searcher/BatchCount" => { + #[allow(non_camel_case_types)] + struct BatchCountSvc(pub Arc); + impl< + T: Searcher, + > tonic::server::UnaryService + for BatchCountSvc { + type Response = super::BatchCountResponse; + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { + ::batch_count(&inner, request).await + }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = BatchCountSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } "/komp_ac.search.Searcher/Count" => { #[allow(non_camel_case_types)] struct CountSvc(pub Arc); diff --git a/komp-app/src/grpc.rs b/komp-app/src/grpc.rs index 43615b62..6908fae5 100644 --- a/komp-app/src/grpc.rs +++ b/komp-app/src/grpc.rs @@ -1244,6 +1244,34 @@ impl GrpcClient { self.search_client.get_filtered_row(request).await } + pub async fn count_table_rows_batch( + &mut self, + profile_name: String, + rows: Vec<(String, String, String)>, + ) -> Vec> { + let mut counts = Vec::with_capacity(rows.len()); + for chunk in rows.chunks(64) { + let result = async { + let request = self.authenticated_request(common::proto::komp_ac::search::BatchCountRequest { + requests: chunk.iter().map(|(table, column, value)| SearchRequest { + profile_name: profile_name.clone(), + table_name: Some(table.clone()), + must: vec![ColumnConstraint { + column: column.clone(), query: value.clone(), mode: MatchMode::Exact as i32, + }], + ..Default::default() + }).collect(), + })?; + self.search_client.batch_count(request).await + }.await; + match result { + Ok(results) => counts.extend(results), + Err(error) => counts.extend((0..chunk.len()).map(|_| Err(error.to_string()))), + } + } + counts + } + pub async fn count_table_rows( &mut self, profile_name: String, diff --git a/komp-app/src/search.rs b/komp-app/src/search.rs index 0370d916..21a145c8 100644 --- a/komp-app/src/search.rs +++ b/komp-app/src/search.rs @@ -28,7 +28,54 @@ impl SearchGrpc { Ok(self.client.get_filtered_row(request).await?.into_inner()) } + pub async fn batch_count( + &mut self, + request: Request, + ) -> Result>> { + let expected = request.get_ref().requests.len(); + let response = self.client.batch_count(request).await?.into_inner(); + decode_batch_counts(response, expected) + } + pub async fn count(&mut self, request: Request) -> Result { Ok(self.client.count(request).await?.into_inner()) } } + +fn decode_batch_counts( + response: common::proto::komp_ac::search::BatchCountResponse, + expected: usize, +) -> Result>> { + use common::proto::komp_ac::search::batch_count_result::Outcome; + anyhow::ensure!(response.results.len() == expected, "Server returned an incomplete count batch"); + Ok(response.results.into_iter().map(|result| match result.outcome { + Some(Outcome::Count(count)) => Ok(count), + Some(Outcome::Error(error)) => Err(error), + None => Err("Server omitted a count result".to_string()), + }).collect()) +} + +#[cfg(test)] +mod batch_count_tests { + use super::*; + use common::proto::komp_ac::search::{BatchCountResponse, BatchCountResult, batch_count_result::Outcome}; + + #[test] + fn batch_counts_preserve_zero_errors_and_missing_results() { + let response = BatchCountResponse { results: vec![ + BatchCountResult { outcome: Some(Outcome::Count(0)) }, + BatchCountResult { outcome: Some(Outcome::Error("denied".into())) }, + BatchCountResult { outcome: None }, + BatchCountResult { outcome: Some(Outcome::Count(7)) }, + ] }; + let counts = decode_batch_counts(response, 4).unwrap(); + assert_eq!(counts[0], Ok(0)); + assert_eq!(counts[1], Err("denied".into())); + assert!(counts[2].is_err()); + assert_eq!(counts[3], Ok(7)); + assert!(decode_batch_counts(BatchCountResponse { results: Vec::new() }, 1).is_err()); + assert!(decode_batch_counts(BatchCountResponse { results: vec![ + BatchCountResult { outcome: Some(Outcome::Count(0)) }, + ] }, 0).is_err()); + } +} diff --git a/search/src/lib.rs b/search/src/lib.rs index 575a9ee4..4c7e3056 100644 --- a/search/src/lib.rs +++ b/search/src/lib.rs @@ -1988,6 +1988,13 @@ impl Searcher for SearcherService { Err(Status::unimplemented("Filtered row loading requires the authorized full-text search service")) } + async fn batch_count( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("Batch counts require the authorized full-text search service")) + } + async fn count( &self, request: Request, diff --git a/server b/server index b085516e..b6092870 160000 --- a/server +++ b/server @@ -1 +1 @@ -Subproject commit b085516e17a37f4f8448e7022dd33183860f9775 +Subproject commit b6092870b17580ec175444b1aec07195f5f5f21e