Use a read-only MetastoreReadService trait for read routing

Replace the full `MetastoreService` implementation on
`ReadReplicaRoutingMetastore` (which forced ~45 delegating methods, most
of them writes) with a narrow, read-only `MetastoreReadService` trait —
the read-only subset of the metastore RPCs (`index_metadata`,
`list_indexes_metadata`, `list_splits`, `list_metrics_splits`,
`list_sketch_splits`).

- `MetastoreServiceClient` implements `MetastoreReadService`;
  `MetastoreReadServiceClient = Arc<dyn MetastoreReadService>`.
- `ReadReplicaRoutingMetastore` now implements only the 5-method trait, so
  writes are excluded at the type level rather than delegated.
- The search and DataFusion read paths take `MetastoreReadServiceClient` /
  `&dyn MetastoreReadService`; `list_parquet_splits_*` take
  `&dyn MetastoreReadService`. `single_node_search` keeps its concrete
  `MetastoreServiceClient` parameter and adapts internally.

Addresses review feedback that the wrapper should expose a Go-style
read-only interface instead of reimplementing the whole service.

Author: shuheiktgw

quickwit/quickwit-datafusion/src/sources/metrics/index_resolver.rs

@@ -25,11 +25,11 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use datafusion::error::Result as DFResult;
 use quickwit_common::uri::Uri;
-use quickwit_metastore::{IndexMetadataResponseExt, ListIndexesMetadataResponseExt};
-use quickwit_parquet_engine::split::ParquetSplitKind;
-use quickwit_proto::metastore::{
-    IndexMetadataRequest, ListIndexesMetadataRequest, MetastoreService, MetastoreServiceClient,
+use quickwit_metastore::{
+    IndexMetadataResponseExt, ListIndexesMetadataResponseExt, MetastoreReadServiceClient,
 };
+use quickwit_parquet_engine::split::ParquetSplitKind;
+use quickwit_proto::metastore::{IndexMetadataRequest, ListIndexesMetadataRequest};
 use tracing::debug;
 
 use super::metastore_provider::MetastoreSplitProvider;
@@ -59,11 +59,11 @@ pub trait MetricsIndexResolver: Send + Sync + std::fmt::Debug {
 /// storage lazily on first read.
 #[derive(Clone)]
 pub struct MetastoreIndexResolver {
-    metastore: MetastoreServiceClient,
+    metastore: MetastoreReadServiceClient,
 }
 
 impl MetastoreIndexResolver {
-    pub fn new(metastore: MetastoreServiceClient) -> Self {
+    pub fn new(metastore: MetastoreReadServiceClient) -> Self {
         Self { metastore }
     }
 }
@@ -85,7 +85,6 @@ impl MetricsIndexResolver for MetastoreIndexResolver {
 
         let response = self
             .metastore
-            .clone()
             .index_metadata(IndexMetadataRequest::for_index_id(index_name.to_string()))
             .await
             .map_err(|err| datafusion::error::DataFusionError::External(Box::new(err)))?;
@@ -111,7 +110,6 @@ impl MetricsIndexResolver for MetastoreIndexResolver {
     async fn list_index_names(&self) -> DFResult<Vec<String>> {
         let response = self
             .metastore
-            .clone()
             .list_indexes_metadata(ListIndexesMetadataRequest::all())
             .await
             .map_err(|err| datafusion::error::DataFusionError::External(Box::new(err)))?;

quickwit/quickwit-datafusion/src/sources/metrics/metastore_provider.rs

@@ -18,9 +18,10 @@ use std::ops::Bound;
 
 use async_trait::async_trait;
 use datafusion::error::Result as DFResult;
-use quickwit_metastore::{ListParquetSplitsQuery, list_parquet_splits_paginated};
+use quickwit_metastore::{
+    ListParquetSplitsQuery, MetastoreReadServiceClient, list_parquet_splits_paginated,
+};
 use quickwit_parquet_engine::split::{ParquetSplitKind, ParquetSplitMetadata};
-use quickwit_proto::metastore::MetastoreServiceClient;
 use quickwit_proto::types::IndexUid;
 use tracing::{debug, instrument};
 
@@ -30,14 +31,14 @@ use super::table_provider::MetricsSplitProvider;
 /// `MetricsSplitProvider` backed by the Quickwit metastore RPC.
 #[derive(Debug, Clone)]
 pub struct MetastoreSplitProvider {
-    metastore: MetastoreServiceClient,
+    metastore: MetastoreReadServiceClient,
     index_uid: IndexUid,
     split_kind: ParquetSplitKind,
 }
 
 impl MetastoreSplitProvider {
     pub fn new(
-        metastore: MetastoreServiceClient,
+        metastore: MetastoreReadServiceClient,
         index_uid: IndexUid,
         split_kind: ParquetSplitKind,
     ) -> Self {
@@ -64,10 +65,13 @@ impl MetricsSplitProvider for MetastoreSplitProvider {
     )]
     async fn list_splits(&self, query: &MetricsSplitQuery) -> DFResult<Vec<ParquetSplitMetadata>> {
         let metastore_query = to_metastore_query(&self.index_uid, query);
-        let records =
-            list_parquet_splits_paginated(self.metastore.clone(), self.split_kind, metastore_query)
-                .await
-                .map_err(|err| datafusion::error::DataFusionError::External(Box::new(err)))?;
+        let records = list_parquet_splits_paginated(
+            self.metastore.as_ref(),
+            self.split_kind,
+            metastore_query,
+        )
+        .await
+        .map_err(|err| datafusion::error::DataFusionError::External(Box::new(err)))?;
         let splits: Vec<ParquetSplitMetadata> =
             records.into_iter().map(|record| record.metadata).collect();
 

quickwit/quickwit-datafusion/src/sources/metrics/mod.rs

@@ -44,8 +44,9 @@ use quickwit_common::{is_metrics_index, is_parquet_pipeline_index, is_sketches_i
 use quickwit_df_core::{
     QuickwitRuntimePlugin, QuickwitRuntimeRegistration, QuickwitSubstraitConsumerExt,
 };
+use quickwit_metastore::MetastoreReadServiceClient;
 use quickwit_parquet_engine::split::ParquetSplitKind;
-use quickwit_proto::metastore::{MetastoreError, MetastoreServiceClient};
+use quickwit_proto::metastore::MetastoreError;
 
 use self::factory::{METRICS_FILE_TYPE, MetricsTableProviderFactory, SKETCHES_FILE_TYPE};
 use self::index_resolver::{MetastoreIndexResolver, MetricsIndexResolver};
@@ -84,7 +85,7 @@ pub struct MetricsDataSource {
 
 impl MetricsDataSource {
     /// Create a production `MetricsDataSource` backed by the metastore.
-    pub fn new(metastore: MetastoreServiceClient) -> Self {
+    pub fn new(metastore: MetastoreReadServiceClient) -> Self {
         Self {
             index_resolver: Arc::new(MetastoreIndexResolver::new(metastore)),
         }

quickwit/quickwit-datafusion/tests/distributed.rs

@@ -69,7 +69,7 @@ async fn test_distributed_tasks_not_shuffles() {
         .await;
     }
 
-    let source = Arc::new(MetricsDataSource::new(metastore));
+    let source = Arc::new(MetricsDataSource::new(Arc::new(metastore)));
 
     // Pool starts empty; we populate it once the workers are bound. The resolver
     // reads pool keys lazily at query time, so this ordering is safe.

quickwit/quickwit-datafusion/tests/metrics.rs

@@ -45,7 +45,7 @@ async fn start_sandbox() -> TestSandbox {
 
 /// Build a `DataFusionSessionBuilder` wired to the sandbox's metastore + storage.
 fn session_builder(sandbox: &TestSandbox) -> DataFusionSessionBuilder {
-    let source = Arc::new(MetricsDataSource::new(sandbox.metastore.clone()));
+    let source = Arc::new(MetricsDataSource::new(Arc::new(sandbox.metastore.clone())));
     let schema_source = Arc::clone(&source);
     let registry = Arc::new(QuickwitObjectStoreRegistry::new(
         sandbox.storage_resolver.clone(),

quickwit/quickwit-datafusion/tests/null_columns.rs

@@ -106,7 +106,7 @@ async fn test_null_columns_for_missing_parquet_fields() {
     );
     publish_split(&metastore, &index_uid, data_dir.path(), "wide", &batch_b).await;
 
-    let source = Arc::new(MetricsDataSource::new(metastore));
+    let source = Arc::new(MetricsDataSource::new(Arc::new(metastore)));
     let schema_source = Arc::clone(&source);
     let registry = Arc::new(QuickwitObjectStoreRegistry::new(
         sandbox.storage_resolver.clone(),

quickwit/quickwit-datafusion/tests/sketches.rs

@@ -36,7 +36,7 @@ async fn start_sandbox() -> TestSandbox {
 }
 
 fn session_builder(sandbox: &TestSandbox) -> DataFusionSessionBuilder {
-    let source = Arc::new(MetricsDataSource::new(sandbox.metastore.clone()));
+    let source = Arc::new(MetricsDataSource::new(Arc::new(sandbox.metastore.clone())));
     let schema_source = Arc::clone(&source);
     let registry = Arc::new(QuickwitObjectStoreRegistry::new(
         sandbox.storage_resolver.clone(),

quickwit/quickwit-index-management/src/parquet_garbage_collection.rs

@@ -177,7 +177,7 @@ async fn list_parquet_splits(
     let kind = parquet_split_kind_for_index(index_uid);
     protect_future(
         progress_opt,
-        list_parquet_splits_paginated(metastore.clone(), kind, query),
+        list_parquet_splits_paginated(metastore, kind, query),
     )
     .await
     .context("failed to list parquet splits")

quickwit/quickwit-indexing/src/actors/parquet_pipeline/parquet_merge_pipeline.rs

@@ -78,7 +78,7 @@ async fn fetch_published_parquet_splits_paginated(
     };
     let query = quickwit_metastore::ListParquetSplitsQuery::for_index(index_uid.clone())
         .retain_immature(OffsetDateTime::now_utc());
-    let records = list_parquet_splits_paginated(metastore, kind, query).await?;
+    let records = list_parquet_splits_paginated(&metastore, kind, query).await?;
     debug!(
         num_splits = records.len(),
         "fetched published parquet splits for merge planning"

quickwit/quickwit-integration-tests/src/tests/metrics_distributed_tests.rs

@@ -323,7 +323,7 @@ async fn test_null_columns_for_missing_parquet_fields() {
     publish_split(&metastore, &index_uid, data_dir.path(), "wide", &batch_b).await;
 
     let storage_resolver = quickwit_storage::StorageResolver::unconfigured();
-    let source = Arc::new(MetricsDataSource::new(metastore));
+    let source = Arc::new(MetricsDataSource::new(Arc::new(metastore)));
     let schema_source = Arc::clone(&source);
     let registry = Arc::new(QuickwitObjectStoreRegistry::new(storage_resolver));
     let builder = DataFusionSessionBuilder::new()