batch count

This commit is contained in:
Priec
2026-09-05 09:16:27 +02:00
parent 52cfbd7070
commit b4fdf706ef
8 changed files with 208 additions and 2 deletions

2
client

Submodule client updated: efe9dfe05b...ede3c38af1

View File

@@ -12,6 +12,8 @@ service Searcher {
// Returns live authorized row data and one-based filtered navigation bounds. // 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. // A missing position returns NOT_FOUND. Search and row reads are not a snapshot.
rpc GetFilteredRow(SearchRequest) returns (komp_ac.tables_data.GetTableDataResponse); 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); rpc Count(SearchRequest) returns (SearchCountResponse);
} }
@@ -127,3 +129,18 @@ message SearchResponse {
} }
repeated Hit hits = 1; 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;
}

Binary file not shown.

View File

@@ -150,6 +150,35 @@ pub mod search_response {
} }
} }
#[derive(serde::Serialize, serde::Deserialize)] #[derive(serde::Serialize, serde::Deserialize)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct BatchCountRequest {
#[prost(message, repeated, tag = "1")]
pub requests: ::prost::alloc::vec::Vec<SearchRequest>,
}
#[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<batch_count_result::Outcome>,
}
/// 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<BatchCountResult>,
}
#[derive(serde::Serialize, serde::Deserialize)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
#[repr(i32)] #[repr(i32)]
pub enum MatchMode { pub enum MatchMode {
@@ -379,6 +408,31 @@ pub mod searcher_client {
.insert(GrpcMethod::new("komp_ac.search.Searcher", "GetFilteredRow")); .insert(GrpcMethod::new("komp_ac.search.Searcher", "GetFilteredRow"));
self.inner.unary(req, path, codec).await 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<super::BatchCountRequest>,
) -> std::result::Result<
tonic::Response<super::BatchCountResponse>,
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( pub async fn count(
&mut self, &mut self,
request: impl tonic::IntoRequest<super::SearchRequest>, request: impl tonic::IntoRequest<super::SearchRequest>,
@@ -433,6 +487,14 @@ pub mod searcher_server {
tonic::Response<super::super::tables_data::GetTableDataResponse>, tonic::Response<super::super::tables_data::GetTableDataResponse>,
tonic::Status, tonic::Status,
>; >;
/// Up to 64 independent counts. Results preserve request order; errors are per item.
async fn batch_count(
&self,
request: tonic::Request<super::BatchCountRequest>,
) -> std::result::Result<
tonic::Response<super::BatchCountResponse>,
tonic::Status,
>;
async fn count( async fn count(
&self, &self,
request: tonic::Request<super::SearchRequest>, request: tonic::Request<super::SearchRequest>,
@@ -603,6 +665,51 @@ pub mod searcher_server {
}; };
Box::pin(fut) Box::pin(fut)
} }
"/komp_ac.search.Searcher/BatchCount" => {
#[allow(non_camel_case_types)]
struct BatchCountSvc<T: Searcher>(pub Arc<T>);
impl<
T: Searcher,
> tonic::server::UnaryService<super::BatchCountRequest>
for BatchCountSvc<T> {
type Response = super::BatchCountResponse;
type Future = BoxFuture<
tonic::Response<Self::Response>,
tonic::Status,
>;
fn call(
&mut self,
request: tonic::Request<super::BatchCountRequest>,
) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move {
<T as Searcher>::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" => { "/komp_ac.search.Searcher/Count" => {
#[allow(non_camel_case_types)] #[allow(non_camel_case_types)]
struct CountSvc<T: Searcher>(pub Arc<T>); struct CountSvc<T: Searcher>(pub Arc<T>);

View File

@@ -1244,6 +1244,34 @@ impl GrpcClient {
self.search_client.get_filtered_row(request).await 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<std::result::Result<u64, String>> {
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( pub async fn count_table_rows(
&mut self, &mut self,
profile_name: String, profile_name: String,

View File

@@ -28,7 +28,54 @@ impl SearchGrpc {
Ok(self.client.get_filtered_row(request).await?.into_inner()) Ok(self.client.get_filtered_row(request).await?.into_inner())
} }
pub async fn batch_count(
&mut self,
request: Request<common::proto::komp_ac::search::BatchCountRequest>,
) -> Result<Vec<std::result::Result<u64, String>>> {
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<SearchRequest>) -> Result<SearchCountResponse> { pub async fn count(&mut self, request: Request<SearchRequest>) -> Result<SearchCountResponse> {
Ok(self.client.count(request).await?.into_inner()) Ok(self.client.count(request).await?.into_inner())
} }
} }
fn decode_batch_counts(
response: common::proto::komp_ac::search::BatchCountResponse,
expected: usize,
) -> Result<Vec<std::result::Result<u64, String>>> {
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());
}
}

View File

@@ -1988,6 +1988,13 @@ impl Searcher for SearcherService {
Err(Status::unimplemented("Filtered row loading requires the authorized full-text search service")) Err(Status::unimplemented("Filtered row loading requires the authorized full-text search service"))
} }
async fn batch_count(
&self,
_request: Request<common::proto::komp_ac::search::BatchCountRequest>,
) -> Result<Response<common::proto::komp_ac::search::BatchCountResponse>, Status> {
Err(Status::unimplemented("Batch counts require the authorized full-text search service"))
}
async fn count( async fn count(
&self, &self,
request: Request<SearchRequest>, request: Request<SearchRequest>,

2
server

Submodule server updated: b085516e17...b6092870b1