Decouple legacy StoreHub client from new Rerun Cloud client (#11050)

* `re_grpc_client` is the client code for `re_grpc_server`, the legacy
gRPC wrapper around the StoreHub.
* `re_cloud_client` is the official gRPC client for the Rerun Cloud
platform.

The name will probably change later -- that doesn't really matter. What
matters is that we decouple these two as they are completely unrelated.

Author: teh-cmc

ARCHITECTURE.md

@@ -98,11 +98,11 @@ Of course, this will only take us so far. In the future we plan on caching queri
 Here is an overview of the crates included in the project:
 
 <picture>
-  <img src="https://static.rerun.io/crates/9f89376256c984dd7999ec5a4e745889eb1a9c5a/full.png" alt="">
-  <source media="(max-width: 480px)" srcset="https://static.rerun.io/crates/9f89376256c984dd7999ec5a4e745889eb1a9c5a/480w.png">
-  <source media="(max-width: 768px)" srcset="https://static.rerun.io/crates/9f89376256c984dd7999ec5a4e745889eb1a9c5a/768w.png">
-  <source media="(max-width: 1024px)" srcset="https://static.rerun.io/crates/9f89376256c984dd7999ec5a4e745889eb1a9c5a/1024w.png">
-  <source media="(max-width: 1200px)" srcset="https://static.rerun.io/crates/9f89376256c984dd7999ec5a4e745889eb1a9c5a/1200w.png">
+  <img src="https://static.rerun.io/crates/1147f3775a6432e22c5015276d0938e9411d54a3/full.png" alt="">
+  <source media="(max-width: 480px)" srcset="https://static.rerun.io/crates/1147f3775a6432e22c5015276d0938e9411d54a3/480w.png">
+  <source media="(max-width: 768px)" srcset="https://static.rerun.io/crates/1147f3775a6432e22c5015276d0938e9411d54a3/768w.png">
+  <source media="(max-width: 1024px)" srcset="https://static.rerun.io/crates/1147f3775a6432e22c5015276d0938e9411d54a3/1024w.png">
+  <source media="(max-width: 1200px)" srcset="https://static.rerun.io/crates/1147f3775a6432e22c5015276d0938e9411d54a3/1200w.png">
 </picture>
 
 <!-- !!! IMPORTANT!!!
@@ -196,10 +196,11 @@ Update instructions:
 
 | Crate                | Description                                                       |
 |----------------------|-------------------------------------------------------------------|
+| re_redap_client      | Official client for the Rerun Data Protocol                       |
 | re_data_loader       | Handles loading of Rerun data from file using data loader plugins |
 | re_data_source       | Handles loading of Rerun data from different sources              |
-| re_grpc_client       | Communicate with the Rerun Data Platform over gRPC                |
-| re_grpc_server       | Host an in-memory Storage Node                                    |
+| re_grpc_client       | Client for the legacy StoreHub API                                |
+| re_grpc_server       | Server for the legacy StoreHub API                                |
 | re_web_viewer_server | Serves the Rerun web viewer (Wasm and HTML) over HTTP             |
 
 ### Build support

Cargo.lock

@@ -7515,6 +7515,7 @@ dependencies = [
  "re_log",
  "re_log_encoding",
  "re_log_types",
+ "re_redap_client",
  "re_smart_channel",
  "re_tracing",
  "re_uri",
@@ -7633,10 +7634,10 @@ dependencies = [
  "itertools 0.14.0",
  "log",
  "re_dataframe",
- "re_grpc_client",
  "re_log_encoding",
  "re_log_types",
  "re_protos",
+ "re_redap_client",
  "re_sorbet",
  "re_tuid",
  "re_uri",
@@ -7749,9 +7750,9 @@ dependencies = [
  "re_chunk_store",
  "re_data_source",
  "re_entity_db",
- "re_grpc_client",
  "re_log",
  "re_log_types",
+ "re_redap_client",
  "re_renderer",
  "re_smart_channel",
  "re_tracing",
@@ -7771,33 +7772,22 @@ dependencies = [
 name = "re_grpc_client"
 version = "0.25.0-alpha.1+dev"
 dependencies = [
- "arrow",
  "async-stream",
  "crossbeam",
- "itertools 0.14.0",
- "jiff",
- "re_arrow_util",
- "re_auth",
  "re_chunk",
- "re_error",
  "re_log",
  "re_log_encoding",
  "re_log_types",
- "re_perf_telemetry",
  "re_protos",
  "re_smart_channel",
  "re_sorbet",
  "re_tracing",
  "re_uri",
- "serde",
  "thiserror 1.0.69",
  "tokio",
  "tokio-stream",
  "tonic",
  "tonic-web-wasm-client",
- "tower",
- "tracing",
- "url",
  "wasm-bindgen-futures",
  "web-time",
 ]
@@ -7849,8 +7839,8 @@ version = "0.25.0-alpha.1+dev"
 dependencies = [
  "egui",
  "egui_kittest",
- "re_grpc_client",
  "re_protos",
+ "re_redap_client",
  "re_sdk",
  "re_server",
  "re_uri",
@@ -8146,11 +8136,11 @@ dependencies = [
  "re_component_ui",
  "re_dataframe_ui",
  "re_datafusion",
- "re_grpc_client",
  "re_log",
  "re_log_encoding",
  "re_log_types",
  "re_protos",
+ "re_redap_client",
  "re_sorbet",
  "re_ui",
  "re_uri",
@@ -8160,6 +8150,36 @@ dependencies = [
  "url",
 ]
 
+[[package]]
+name = "re_redap_client"
+version = "0.25.0-alpha.1+dev"
+dependencies = [
+ "arrow",
+ "itertools 0.14.0",
+ "jiff",
+ "re_arrow_util",
+ "re_auth",
+ "re_chunk",
+ "re_error",
+ "re_log",
+ "re_log_encoding",
+ "re_log_types",
+ "re_perf_telemetry",
+ "re_protos",
+ "re_smart_channel",
+ "re_uri",
+ "serde",
+ "thiserror 1.0.69",
+ "tokio",
+ "tokio-stream",
+ "tonic",
+ "tonic-web-wasm-client",
+ "tower",
+ "tracing",
+ "url",
+ "wasm-bindgen-futures",
+]
+
 [[package]]
 name = "re_renderer"
 version = "0.25.0-alpha.1+dev"
@@ -8404,10 +8424,10 @@ dependencies = [
  "re_chunk_store",
  "re_entity_db",
  "re_global_context",
- "re_grpc_client",
  "re_log",
  "re_log_encoding",
  "re_log_types",
+ "re_redap_client",
  "re_renderer",
  "re_tracing",
  "re_types",
@@ -8976,7 +8996,6 @@ dependencies = [
  "re_entity_db",
  "re_error",
  "re_format",
- "re_grpc_client",
  "re_log",
  "re_log_encoding",
  "re_log_types",
@@ -8985,6 +9004,7 @@ dependencies = [
  "re_query",
  "re_recording_panel",
  "re_redap_browser",
+ "re_redap_client",
  "re_renderer",
  "re_selection_panel",
  "re_smart_channel",
@@ -9360,7 +9380,6 @@ dependencies = [
  "re_format",
  "re_format_arrow",
  "re_global_context",
- "re_grpc_client",
  "re_grpc_server",
  "re_log",
  "re_log_encoding",
@@ -9369,6 +9388,7 @@ dependencies = [
  "re_memory",
  "re_perf_telemetry",
  "re_protos",
+ "re_redap_client",
  "re_sdk",
  "re_server",
  "re_smart_channel",
@@ -9463,6 +9483,7 @@ dependencies = [
  "re_memory",
  "re_perf_telemetry",
  "re_protos",
+ "re_redap_client",
  "re_sdk",
  "re_sorbet",
  "re_tuid",

Cargo.toml

@@ -71,6 +71,7 @@ re_dataframe = { path = "crates/store/re_dataframe", version = "=0.25.0-alpha.1"
 re_datafusion = { path = "crates/store/re_datafusion", version = "=0.25.0-alpha.1", default-features = false }
 re_entity_db = { path = "crates/store/re_entity_db", version = "=0.25.0-alpha.1", default-features = false }
 re_format_arrow = { path = "crates/store/re_format_arrow", version = "=0.25.0-alpha.1", default-features = false }
+re_redap_client = { path = "crates/store/re_redap_client", version = "=0.25.0-alpha.1", default-features = false }
 re_grpc_client = { path = "crates/store/re_grpc_client", version = "=0.25.0-alpha.1", default-features = false }
 re_grpc_server = { path = "crates/store/re_grpc_server", version = "=0.25.0-alpha.1", default-features = false }
 re_protos = { path = "crates/store/re_protos", version = "=0.25.0-alpha.1", default-features = false }
@@ -452,7 +453,7 @@ debug = false
 "re_error".debug = true
 "re_format".debug = true
 "re_format_arrow".debug = true
-"re_grpc_client".debug = true
+"re_redap_client".debug = true
 "re_grpc_server".debug = true
 "re_int_histogram".debug = true
 "re_log".debug = true

crates/store/re_data_source/Cargo.toml

@@ -25,6 +25,7 @@ default = []
 [dependencies]
 re_data_loader.workspace = true
 re_error.workspace = true
+re_redap_client.workspace = true
 re_grpc_client.workspace = true
 re_log_encoding = { workspace = true, features = [
   "decoder",

crates/store/re_data_source/src/data_source.rs

@@ -1,5 +1,5 @@
-use re_grpc_client::ConnectionRegistryHandle;
 use re_log_types::{LogMsg, RecordingId};
+use re_redap_client::ConnectionRegistryHandle;
 use re_smart_channel::{Receiver, SmartChannelSource, SmartMessageSource};
 
 use crate::FileContents;
@@ -146,7 +146,7 @@ impl LogDataSource {
     pub fn stream(
         self,
         connection_registry: &ConnectionRegistryHandle,
-        on_ui_cmd: Option<Box<dyn Fn(re_grpc_client::UiCommand) + Send + Sync>>,
+        on_ui_cmd: Option<Box<dyn Fn(re_redap_client::UiCommand) + Send + Sync>>,
         on_msg: Option<Box<dyn Fn() + Send + Sync>>,
     ) -> anyhow::Result<Receiver<LogMsg>> {
         re_tracing::profile_function!();
@@ -253,7 +253,7 @@ impl LogDataSource {
                 let uri_clone = uri.clone();
                 let stream_partition = async move {
                     let client = connection_registry.client(uri_clone.origin.clone()).await?;
-                    re_grpc_client::stream_blueprint_and_partition_from_server(
+                    re_redap_client::stream_blueprint_and_partition_from_server(
                         client, tx, uri_clone, on_ui_cmd, on_msg,
                     )
                     .await
@@ -270,7 +270,7 @@ impl LogDataSource {
                 Ok(rx)
             }
 
-            Self::RedapProxy(uri) => Ok(re_grpc_client::message_proxy::stream(uri, on_msg)),
+            Self::RedapProxy(uri) => Ok(re_grpc_client::stream(uri, on_msg)),
         }
     }
 }

crates/store/re_datafusion/Cargo.toml

@@ -31,7 +31,7 @@ default = []
 [dependencies]
 # Rerun dependencies:
 re_dataframe.workspace = true
-re_grpc_client.workspace = true
+re_redap_client.workspace = true
 re_log_encoding.workspace = true
 re_log_types.workspace = true
 re_protos.workspace = true

crates/store/re_datafusion/examples/catalog.rs

@@ -10,7 +10,7 @@ use re_tuid::Tuid;
 async fn main() -> anyhow::Result<()> {
     let local_addr = "rerun+http:://127.0.0.1:51234";
 
-    let connection_registry = re_grpc_client::ConnectionRegistry::new();
+    let connection_registry = re_redap_client::ConnectionRegistry::new();
 
     let client = connection_registry.client(local_addr.parse()?).await?;
     let mut df_connector = DataFusionConnector::new(client).await?;

crates/store/re_datafusion/src/dataframe_query_common.rs

@@ -1,3 +1,9 @@
+use std::any::Any;
+use std::cmp::Ordering;
+use std::collections::{BTreeMap, BTreeSet};
+use std::str::FromStr as _;
+use std::sync::Arc;
+
 use arrow::array::{
     ArrayRef, DurationNanosecondArray, Int64Array, RecordBatch, StringArray,
     TimestampMicrosecondArray, TimestampMillisecondArray, TimestampNanosecondArray,
@@ -12,25 +18,21 @@ use datafusion::datasource::TableType;
 use datafusion::logical_expr::{Expr, Operator, TableProviderFilterPushDown};
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion::physical_plan::coalesce_batches::CoalesceBatchesExec;
+
 use re_dataframe::external::re_chunk::ChunkId;
 use re_dataframe::external::re_chunk_store::ChunkStore;
 use re_dataframe::{Index, QueryExpression};
-use re_grpc_client::{ConnectionClient, ConnectionRegistryHandle};
 use re_log_encoding::codec::wire::decoder::Decode as _;
 use re_log_types::EntryId;
 use re_log_types::external::re_types_core::Loggable as _;
 use re_protos::cloud::v1alpha1::DATASET_MANIFEST_ID_FIELD_NAME;
 use re_protos::cloud::v1alpha1::ext::{Query, QueryLatestAt, QueryRange};
 use re_protos::cloud::v1alpha1::{GetChunksRequest, GetDatasetSchemaRequest, QueryDatasetRequest};
 use re_protos::common::v1alpha1::ext::ScanParameters;
+use re_redap_client::{ConnectionClient, ConnectionRegistryHandle};
 use re_sorbet::{BatchType, ChunkColumnDescriptors, ColumnKind, ComponentColumnSelector};
 use re_tuid::Tuid;
 use re_uri::Origin;
-use std::any::Any;
-use std::cmp::Ordering;
-use std::collections::{BTreeMap, BTreeSet};
-use std::str::FromStr as _;
-use std::sync::Arc;
 
 /// Sets the size for output record batches in rows. The last batch will likely be smaller.
 /// The default for Data Fusion is 8192, which leads to a 256Kb record batch on average for

crates/store/re_datafusion/src/dataframe_query_provider.rs

@@ -1,7 +1,10 @@
-use crate::dataframe_query_common::{
-    ChunkInfo, align_record_batch_to_schema, compute_partition_stream_chunk_info,
-    prepend_string_column_schema,
-};
+use std::any::Any;
+use std::collections::BTreeMap;
+use std::fmt::Debug;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+
 use arrow::array::{Array, RecordBatch, RecordBatchOptions, StringArray};
 use arrow::compute::SortOptions;
 use arrow::datatypes::{Schema, SchemaRef};
@@ -17,28 +20,28 @@ use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
 use datafusion::physical_plan::{DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties};
 use datafusion::{error::DataFusionError, execution::SendableRecordBatchStream};
 use futures_util::{Stream, StreamExt as _};
+use tokio::runtime::Handle;
+use tokio::sync::Notify;
+use tokio::sync::mpsc::{Receiver, Sender};
+use tokio::task::JoinHandle;
+use tracing::Instrument as _;
+
 use re_dataframe::external::re_chunk::Chunk;
 use re_dataframe::external::re_chunk_store::ChunkStore;
 use re_dataframe::{
     ChunkStoreHandle, Index, QueryCache, QueryEngine, QueryExpression, QueryHandle, StorageEngine,
 };
-use re_grpc_client::ConnectionClient;
 use re_log_types::{ApplicationId, StoreId, StoreInfo, StoreKind, StoreSource};
 use re_protos::cloud::v1alpha1::DATASET_MANIFEST_ID_FIELD_NAME;
 use re_protos::cloud::v1alpha1::GetChunksRequest;
 use re_protos::common::v1alpha1::PartitionId;
+use re_redap_client::ConnectionClient;
 use re_sorbet::{ColumnDescriptor, ColumnSelector};
-use std::any::Any;
-use std::collections::BTreeMap;
-use std::fmt::Debug;
-use std::pin::Pin;
-use std::sync::Arc;
-use std::task::{Context, Poll};
-use tokio::runtime::Handle;
-use tokio::sync::Notify;
-use tokio::sync::mpsc::{Receiver, Sender};
-use tokio::task::JoinHandle;
-use tracing::Instrument as _;
+
+use crate::dataframe_query_common::{
+    ChunkInfo, align_record_batch_to_schema, compute_partition_stream_chunk_info,
+    prepend_string_column_schema,
+};
 
 /// This parameter sets the back pressure that either the streaming provider
 /// can place on the CPU worker thread or the CPU worker thread can place on
@@ -455,7 +458,7 @@ async fn chunk_stream_io_loop(
 
         // Then we need to fully decode these chunks, i.e. both the transport layer (Protobuf)
         // and the app layer (Arrow).
-        let mut chunk_stream = re_grpc_client::get_chunks_response_to_chunk_and_partition_id(
+        let mut chunk_stream = re_redap_client::get_chunks_response_to_chunk_and_partition_id(
             get_chunks_response_stream,
         );