Refactoring and minor bug fix. (#1108)

Simplified the search client pool:
- no async
- no trait
- a sync rwlock that copies modifies and sets.
(The previous implementation held the lock while the client where
created)
- Removed redundant code
- Removed redundant Arc<> over SearchClientPool

Refactor root.rs to encapsulate Partial hit map contract
Added a job trait.

Removing couple in placed requests

Renamed split_metadata -> split_offsets in the proto

Simplified rendez vous hashing code

Removing logic limiting the number of leaf requests
run concurrently in the root.

Refactoring attempt to enforce split offset contract.

This refactoring attempts to remove a bunch of
unwraps justified by non-trivial contracts in the
root code.

Author: fulmicoton

quickwit-proto/proto/search_api.proto

@@ -123,7 +123,7 @@ message LeafSearchRequest {
 
   // Index split ids to apply the query on.
   // This ids are resolved from the index_uri defined in the search_request.
-  repeated SplitIdAndFooterOffsets split_metadata = 4;
+  repeated SplitIdAndFooterOffsets split_offsets = 4;
 
   // `DocMapper` as json serialized trait.
   string doc_mapper = 5;
@@ -204,7 +204,7 @@ message FetchDocsRequest {
   // Split footer offsets. They are required for fetch docs to
   // fetch the document content in two reads, when the footer is not
   // cached.
-  repeated SplitIdAndFooterOffsets split_metadata = 3;
+  repeated SplitIdAndFooterOffsets split_offsets = 3;
 
   // Index URI. The index URI defines the location of the storage that contains the
   // split files.
@@ -261,7 +261,7 @@ message LeafSearchStreamRequest {
 
   // Index split ids to apply the query on.
   // This ids are resolved from the index_uri defined in the stream request.
-  repeated SplitIdAndFooterOffsets split_metadata = 2;
+  repeated SplitIdAndFooterOffsets split_offsets = 2;
 
   // `DocMapper` as json serialized trait.
   string doc_mapper = 5;

quickwit-proto/src/quickwit.rs

@@ -78,7 +78,7 @@ pub struct LeafSearchRequest {
     /// Index split ids to apply the query on.
     /// This ids are resolved from the index_uri defined in the search_request.
     #[prost(message, repeated, tag = "4")]
-    pub split_metadata: ::prost::alloc::vec::Vec<SplitIdAndFooterOffsets>,
+    pub split_offsets: ::prost::alloc::vec::Vec<SplitIdAndFooterOffsets>,
     /// `DocMapper` as json serialized trait.
     #[prost(string, tag = "5")]
     pub doc_mapper: ::prost::alloc::string::String,
@@ -175,7 +175,7 @@ pub struct FetchDocsRequest {
     /// fetch the document content in two reads, when the footer is not
     /// cached.
     #[prost(message, repeated, tag = "3")]
-    pub split_metadata: ::prost::alloc::vec::Vec<SplitIdAndFooterOffsets>,
+    pub split_offsets: ::prost::alloc::vec::Vec<SplitIdAndFooterOffsets>,
     /// Index URI. The index URI defines the location of the storage that contains the
     /// split files.
     #[prost(string, tag = "4")]
@@ -228,7 +228,7 @@ pub struct LeafSearchStreamRequest {
     /// Index split ids to apply the query on.
     /// This ids are resolved from the index_uri defined in the stream request.
     #[prost(message, repeated, tag = "2")]
-    pub split_metadata: ::prost::alloc::vec::Vec<SplitIdAndFooterOffsets>,
+    pub split_offsets: ::prost::alloc::vec::Vec<SplitIdAndFooterOffsets>,
     /// `DocMapper` as json serialized trait.
     #[prost(string, tag = "5")]
     pub doc_mapper: ::prost::alloc::string::String,

quickwit-search/src/client.rs

@@ -22,12 +22,11 @@ use std::net::SocketAddr;
 use std::sync::Arc;
 
 use futures::{StreamExt, TryStreamExt};
-use http::Uri;
 use opentelemetry::global;
 use opentelemetry::propagation::Injector;
 use quickwit_proto::LeafSearchStreamResponse;
 use tokio_stream::wrappers::UnboundedReceiverStream;
-use tonic::transport::{Channel, Endpoint};
+use tonic::transport::Channel;
 use tonic::Request;
 use tracing::*;
 use tracing_opentelemetry::OpenTelemetrySpanExt;
@@ -229,22 +228,3 @@ impl SearchServiceClient {
         }
     }
 }
-
-/// Create a SearchServiceClient with SocketAddr as an argument.
-/// It will try to reconnect to the node automatically.
-pub async fn create_search_service_client(
-    grpc_addr: SocketAddr,
-) -> anyhow::Result<SearchServiceClient> {
-    let uri = Uri::builder()
-        .scheme("http")
-        .authority(grpc_addr.to_string().as_str())
-        .path_and_query("/")
-        .build()?;
-    // Create a channel with connect_lazy to automatically reconnect to the node.
-    let channel = Endpoint::from(uri).connect_lazy();
-    let client = SearchServiceClient::from_grpc_client(
-        quickwit_proto::search_service_client::SearchServiceClient::new(channel),
-        grpc_addr,
-    );
-    Ok(client)
-}