From b4fdf706efc686ed740219b868b4ed160a86b3b5 Mon Sep 17 00:00:00 2001 From: Priec Date: Sat, 5 Sep 2026 09:16:27 +0200 Subject: [PATCH] batch count --- client | 2 +- common/proto/search.proto | 17 +++++ common/src/proto/descriptor.bin | Bin 248415 -> 249265 bytes common/src/proto/komp_ac.search.rs | 107 +++++++++++++++++++++++++++++ komp-app/src/grpc.rs | 28 ++++++++ komp-app/src/search.rs | 47 +++++++++++++ search/src/lib.rs | 7 ++ server | 2 +- 8 files changed, 208 insertions(+), 2 deletions(-) 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 c7a4717337cf2e2d3c5d8cb67e5a79e45fa9d5be..60f31304fb514b35e4b3a25bdc5ae09d9844f8ff 100644 GIT binary patch delta 4681 zcmaJ_OHUlx6|Q^hR&{lC^C}u>npg2OFouaS#&-PRVKcHljt$I!ZM?t?^aMPCW|{^I zkX-i2WFARmM~cKEKOl>2oG238ETSmOC-hK-=>Co9{3G2tCh8lTcy42%0ltWz2a^qxaxal^UvM9 z2<^elSjWkVE1OSte3^+BV!*x`jFr!308Qx=Fqz(M_;(u*NHJJU@uf?j97m3c6A$vm5cLnx+EFEAVj>2uVUq?)40;_p4LuP< zoCW~ohFH$b#}KCh1i2xkL2`DW&RZWwMvtcayl0wb(;1COJyQw*dLyg{1jmsy1_$Q{ z1ig`_W}SvjIgN6!0YDo~XjIi`+B6q7&_+#jBc%d$!MbYFP;k58<+YxoG{&SkRsiUY z$&?NZ1jjM2Hb&4JYY3PI(?#xv(rH}eG=Rv)-2ehDy2z)Z=nfj^QvpD@@q|aEjq{slb~^!ciM0xXG}AH^EI-l-Ffx zRs=w!xG4^3mpwDjuz_~DQC~s`;0E@Uq&OF}D<&HHPj{DnC8^E@?+TBc6R2y}j5!-p zzvi{;COdRC*STr{=v|ktJ`Er^UQd~`0fO9huG*nK-jEIKNFx&6NO*KOH&WTy2;Pml zdJc!PxtWBc9^6c5qrijDQ;Y{)68 z*-RxoI-99EHmSGqEJ~a=IPQj~_V-MyO2p59qaY z19jW_$h4de<~Cn8F137?djNn~vz!zV9B0#}P8{i6K29994UTm4 zqyqK8S}-L>DY!l0l}}NP52b!y6aZ>`$Tb4N@nKrOoMc0d4+H%q0>L!LtpNaSE}>Dr zbKDx(K%3+7NKsDFJTDUvj%Z%$mmZR&IiEGl1O(nZmWk5$0`;l&%%tzZ>r<~upT9@h zFY>qnAdN*%0|<_beDeT7Zc$Gg_RphGOR`(f4FKk5iRT6gDoYu&F+k)Qw8x%MS?V$m zt4GB=;tdA??NJh(0zBdchYhqxe0_Nlp~teHC2a(MoCLSgk;go=p5B?H~TnAT)xjBsRY-938C z^hd$E)L(*dB#m|1MUD|cc&_*KvL`^%*kGcwEt;VtM7EAjJ*VkFZCGEJIZeZB!+b%i zG|lNI7X^UMZt|!A!Euwf3n57!JCD@Y%9u zeQ8=l3R{l;q;vFV%eEZS%`hJ-+gv{o>exOvNCijm1g&Jwlz>M1*a*S1WR?I3U`Of$ zA^^u7Gf#xb7Is5YDnD?_)>p#x#)n(kYxZk!?!IjW&aQkVY)Xru2m*T-wjSRlNWH6^ z9Jk57sG`z-t)DW#LZbpog3*8qUauNv!Mvw(iJ7)~?+l$v+VktGcF_0G_Lz>;_f>mh zrbXG0lwbV;{QXQz6@$NzwTk(GFH|8h(@Xf`44vM>7n!E2o&0e<&2#oxq676oVJ5s) z+Wvg=r+ek%dbn}&yLLGf9jI#2^iUquG$t~%>j$}kDu(RA&~TC!+z-|H#7ql+ct(y^ z{_u<(E&XBcT-6_P&klz_Nc1p}N2)uq)9xRg(W9@PqceK6|3|rMSLp&c!X&*CTsE$N zmnVPgkY7bF&zu!q0x$i15{9mUm$-f6UVB&>)ts0;M2(bJ#h^#&hAIX<%FzcMrVrDl zUJKmfH8gzE!nZM?AH^6S2d_n@Hio!$UW+;!6YY&J`qu&%dlnO(3EVfr`Z-2CggQH98ip-LV-_jzu3m7*b^e_ZwmTilb!V z`i;O_UW+JM&kEdch5RK}iSh%6`CH-93l1|S@tsgj#z?miL6ZcG@cmBUZ!=Ae=pq8f z18lgk{-VmoHW7@uSSG-jOLq~Vrb<;4xNq^lAUNR~c)k@L{WYpo4Q2FB$Uz+w2yFo* n)nSLM=*v5S*N{>I1oxe2?~Ng%yb}ZGzW%;bo_^nHZ~y+kIOUZA delta 3851 zcmYjT%T63e6z#rU-P6;9!89;4!0;L{*g#~!fMvj7f#a79IOe?|GPYx|Y-5{;fC$nk zij-wGQHbPE$i|5*@d^2iEb<#!?XNL1rAZ{Rnx7g zD)Je)KtV1=Y^!Qakq6XuGm0V#+*s8*buNB;feTJ;SRb*ROHS;?^tGk{z!V2HDS}*l zxhq8p^y2F?T90DY;4~UNj-er}(|W{ud3vll05Ii)LW-c5FAk&#a{2K`LW5(0rp3e& z7J@iMpcU%6ql8mN#^4nO?+6dJBHa^Dj-(jWr-*G0y(t2%IC7VxS5B;&oi-i4;t-p| zc9v5PM_-~QrZDgsg(*tlS;{t~5PGFzqwrvAp`w^v@LDX7lWED;r8dx7@&%zG!d7Y# z6DdYfiXhUOyPP73B(1gS2?Uka=8{eZt?h8SOfq~{N7$^@b!*F#3;^`X5|n9lS#kk_ zTv>8)xrm*T3jk=HhL$KmHcB!;;B_Jyt|=y>+qr7OV2SRqMTem$qDR62fL@P;VdkSp z!T^F?55izMFIK(IS4Pf*TyI!Nw4TR3>LWc<3IKY2q6Y-ezN{V`wn4A2AulwT`lZ&M zPOIP2III4wDK2cF^-FP8tgbsZO&G{O_O6FTt*1DQ0Wxg`Ko|p*(cJ=q<3L!QBIpg& z#zKQ>P^zJH7=sc95Zk00K%fmuQxtD~G9+CA!f`{E$K@W9roaZ?kW83jaKlFFfp)`2 z$z{D^qx9I*Mu|}zmKrNkpbb-)NIqPKVJR|<;0;TW6~{F~W<>xrN{ixyHWHeFh7Gil zI(-Q#MM3P1+T;SzMs0Gqy1Y@HT;j7Qc%w3MeynaecT96w{Z?4gMfSNjW0Exh^u{RA z(F4J8EMs~D1i3NE+UGisQ>{4i>z-fD!TS0lzHhdb0ZjKP(F@8D^}de6-#nREJ$9a$6f^L8EE}ImILBGi&x^|F z9A_m*Ab8GZ^~=dNhwyqDHt7C`9Q}ff8vw#skT8JYxFF*O1i1ydc|tyE zkvcF8*gqi5&7#Z=5L6a(W@CWZH7@D70fNe6n|WA6PG(6q900T>dpMu4B-a;gpe@Pu z6*5?+D@HLP4~H+?!#xb`vJ5Rep|V_Rm&18umOHy7bI?{OjZ0}{p4m{}Z7VhdUZ0iN zye+UnWTnZxrXl%B?5)ZJ0tA&+8w%fat5${gXEip@2|S^)YNi$SL(FTsnq15PB5PJ8 zU|Y`YE^H84E2Kg5wO6x-YJPU{of1+e_SVmSZ=%1Fbs0$f2gZ7XSz915)_HBQX81&S z=Daj(#!Y(W>kklLe zcmljIHki1#FU-&}Vp~hAp3`isUOBJLoMz$m%6vhpEYIntBn5!VZpx?t!EsZz3>Qt?_+c+Dn|iBF5S-AY72EGJ1QGhM;Y@kBz8ct82Rtu z@wy}|n0HmZHS<#MUZ8VEyHQQW4*D*tp3)Kdo+?>0FZW)eqRJoO@8t>=4E`S0D&+(2 zt0rsahj9M_ofmsQ*HE#uKW?XQBzo-7f$D3TZl8I-{p$7R{?2zV+qcfXZ>8zvKvk0F zck$OqA{ZRE=Gw=K11v%dM!wYh}^N0046@SP*JM8VVdYH!} z)nV=8~_OtQ6vZXB{}oP13@RL WF(1c3P)YKIpJ&Q+K2!F#|M(wl-|, +} +#[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