Replace QueryStorage grpc with QueryContext

crates/admin/Cargo.toml

@@ -10,7 +10,6 @@ publish = false
 [features]
 default = ["replicated-loglet", "serve-web-ui"]
 clients = []
-servers = []
 options_schema = ["restate-service-client/options_schema"]
 memory-loglet = ["restate-bifrost/memory-loglet"]
 replicated-loglet = ["restate-bifrost/replicated-loglet"]
@@ -23,15 +22,16 @@ restate-core = { workspace = true }
 restate-errors = { workspace = true }
 restate-fs-util = { workspace = true }
 restate-futures-util = { workspace = true }
+restate-metadata-store = { workspace = true }
 restate-service-client = { workspace = true }
 restate-service-protocol = { workspace = true, features = ["discovery"] }
+restate-storage-query-datafusion = { workspace = true }
 restate-types = { workspace = true, features = ["schemars"] }
 restate-wal-protocol = { workspace = true }
 restate-web-ui = { workspace = true, optional = true }
 
 anyhow = { workspace = true }
 arc-swap = { workspace = true }
-arrow-flight = { workspace = true }
 axum = { workspace = true }
 bytes = { workspace = true }
 bytestring = { workspace = true }

crates/admin/build.rs

@@ -17,7 +17,6 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     tonic_build::configure()
         .bytes(["."])
         .file_descriptor_set_path(out_dir.join("cluster_ctrl_svc_descriptor.bin"))
-        .server_mod_attribute("cluster_ctrl", "#[cfg(feature = \"servers\")]")
         .client_mod_attribute("cluster_ctrl", "#[cfg(feature = \"clients\")]")
         // allow older protobuf compiler to be used
         .protoc_arg("--experimental_allow_proto3_optional")

crates/node/src/network_server/handler/cluster_ctrl.rs → crates/admin/src/cluster_controller/grpc_svc_handler.rs

@@ -11,20 +11,11 @@
 use std::time::Duration;
 
 use bytes::{Bytes, BytesMut};
-use restate_core::MetadataWriter;
 use tonic::{async_trait, Request, Response, Status};
 use tracing::{debug, info};
 
-use restate_admin::cluster_controller::protobuf::cluster_ctrl_svc_server::ClusterCtrlSvc;
-use restate_admin::cluster_controller::protobuf::{
-    ClusterStateRequest, ClusterStateResponse, CreatePartitionSnapshotRequest,
-    CreatePartitionSnapshotResponse, DescribeLogRequest, DescribeLogResponse, FindTailRequest,
-    FindTailResponse, ListLogsRequest, ListLogsResponse, ListNodesRequest, ListNodesResponse,
-    SealAndExtendChainRequest, SealAndExtendChainResponse, SealedSegment, TailState,
-    TrimLogRequest,
-};
-use restate_admin::cluster_controller::ClusterControllerHandle;
 use restate_bifrost::{Bifrost, BifrostAdmin, Error as BiforstError};
+use restate_core::MetadataWriter;
 use restate_metadata_store::MetadataStoreClient;
 use restate_types::identifiers::PartitionId;
 use restate_types::logs::metadata::{Logs, ProviderKind, SegmentIndex};
@@ -34,22 +25,36 @@ use restate_types::nodes_config::NodesConfiguration;
 use restate_types::storage::{StorageCodec, StorageEncode};
 use restate_types::{Version, Versioned};
 
-use crate::network_server::ClusterControllerDependencies;
+use crate::cluster_controller::protobuf::cluster_ctrl_svc_server::ClusterCtrlSvc;
+use crate::cluster_controller::protobuf::{
+    ClusterStateRequest, ClusterStateResponse, CreatePartitionSnapshotRequest,
+    CreatePartitionSnapshotResponse, DescribeLogRequest, DescribeLogResponse, FindTailRequest,
+    FindTailResponse, ListLogsRequest, ListLogsResponse, ListNodesRequest, ListNodesResponse,
+    SealAndExtendChainRequest, SealAndExtendChainResponse, SealedSegment, TailState,
+    TrimLogRequest,
+};
+
+use super::ClusterControllerHandle;
 
-pub struct ClusterCtrlSvcHandler {
+pub(crate) struct ClusterCtrlSvcHandler {
     metadata_store_client: MetadataStoreClient,
     controller_handle: ClusterControllerHandle,
-    bifrost_handle: Bifrost,
+    bifrost: Bifrost,
     metadata_writer: MetadataWriter,
 }
 
 impl ClusterCtrlSvcHandler {
-    pub fn new(admin_deps: ClusterControllerDependencies) -> Self {
+    pub fn new(
+        controller_handle: ClusterControllerHandle,
+        metadata_store_client: MetadataStoreClient,
+        bifrost: Bifrost,
+        metadata_writer: MetadataWriter,
+    ) -> Self {
         Self {
-            controller_handle: admin_deps.cluster_controller_handle,
-            metadata_store_client: admin_deps.metadata_store_client,
-            bifrost_handle: admin_deps.bifrost_handle,
-            metadata_writer: admin_deps.metadata_writer,
+            controller_handle,
+            metadata_store_client,
+            bifrost,
+            metadata_writer,
         }
     }
 
@@ -107,7 +112,7 @@ impl ClusterCtrlSvc for ClusterCtrlSvcHandler {
             .clone();
 
         let (trim_point, nodes_config) = tokio::join!(
-            self.bifrost_handle.get_trim_point(log_id),
+            self.bifrost.get_trim_point(log_id),
             self.metadata_store_client
                 .get::<NodesConfiguration>(NODES_CONFIG_KEY.clone()),
         );
@@ -207,7 +212,7 @@ impl ClusterCtrlSvc for ClusterCtrlSvcHandler {
         request: Request<SealAndExtendChainRequest>,
     ) -> Result<Response<SealAndExtendChainResponse>, Status> {
         let admin = BifrostAdmin::new(
-            &self.bifrost_handle,
+            &self.bifrost,
             &self.metadata_writer,
             &self.metadata_store_client,
         );
@@ -250,7 +255,7 @@ impl ClusterCtrlSvc for ClusterCtrlSvcHandler {
         let log_id: LogId = request.log_id.into();
 
         let admin = BifrostAdmin::new(
-            &self.bifrost_handle,
+            &self.bifrost,
             &self.metadata_writer,
             &self.metadata_store_client,
         );

crates/admin/src/cluster_controller/mod.rs

@@ -9,6 +9,7 @@
 // by the Apache License, Version 2.0.
 
 pub mod cluster_state_refresher;
+pub mod grpc_svc_handler;
 mod logs_controller;
 mod observed_cluster_state;
 pub mod protobuf;