minor refact

Author: fulmicoton-dd

quickwit/quickwit-cli/src/tool.rs

@@ -37,6 +37,7 @@ use quickwit_config::{
     TransformConfig, VecSourceParams,
 };
 use quickwit_index_management::{IndexService, clear_cache_directory};
+use quickwit_indexing::BoxedPipelineHandle;
 use quickwit_indexing::actors::{IndexingService, MergePipeline, MergeSchedulerService};
 use quickwit_indexing::models::{
     DetachIndexingPipeline, DetachMergePipeline, IndexingStatistics, SpawnPipeline,
@@ -750,7 +751,7 @@ async fn extract_split_cli(args: ExtractSplitArgs) -> anyhow::Result<()> {
 /// Starts a tokio task that displays the indexing statistics
 /// every once in awhile.
 pub async fn start_statistics_reporting_loop(
-    pipeline_handle: quickwit_indexing::BoxPipelineHandle,
+    pipeline_handle: BoxedPipelineHandle,
     is_stdin: bool,
 ) -> anyhow::Result<IndexingStatistics> {
     let mut stdout_handle = stdout();

quickwit/quickwit-indexing/src/actors/indexing_service.rs

@@ -87,7 +87,7 @@ struct MergePipelineHandle {
     handle: ActorHandle<MergePipeline>,
 }
 
-pub type BoxPipelineHandle = Box<dyn PipelineHandle>;
+pub type BoxedPipelineHandle = Box<dyn PipelineHandle>;
 
 /// The indexing service is (single) actor service running on indexer and in charge
 /// of executing the indexing plans received from the control plane.
@@ -105,7 +105,7 @@ pub struct IndexingService {
     merge_scheduler_service: Mailbox<MergeSchedulerService>,
     pub(crate) ingester_pool: IngesterPool,
     pub(crate) storage_resolver: StorageResolver,
-    indexing_pipelines: HashMap<PipelineUid, BoxPipelineHandle>,
+    indexing_pipelines: HashMap<PipelineUid, BoxedPipelineHandle>,
     counters: IndexingServiceCounters,
     local_split_store: Arc<IndexingSplitCache>,
     pub(crate) max_concurrent_split_uploads: usize,
@@ -182,7 +182,7 @@ impl IndexingService {
     async fn detach_indexing_pipeline(
         &mut self,
         pipeline_uid: &PipelineUid,
-    ) -> Result<BoxPipelineHandle, IndexingError> {
+    ) -> Result<BoxedPipelineHandle, IndexingError> {
         let pipeline_handle = self
             .indexing_pipelines
             .remove(pipeline_uid)
@@ -326,7 +326,7 @@ impl IndexingService {
         source_config: SourceConfig,
         immature_splits_opt: Option<Vec<SplitMetadata>>,
         params_fingerprint: u64,
-    ) -> Result<BoxPipelineHandle, IndexingError> {
+    ) -> Result<BoxedPipelineHandle, IndexingError> {
         let pipeline_uid_str = indexing_pipeline_id.pipeline_uid.to_string();
         let indexing_directory = temp_dir::Builder::default()
             .join(&indexing_pipeline_id.index_uid.index_id)
@@ -916,7 +916,7 @@ impl Handler<ObservePipeline> for IndexingService {
 
 #[async_trait]
 impl Handler<DetachIndexingPipeline> for IndexingService {
-    type Reply = Result<BoxPipelineHandle, IndexingError>;
+    type Reply = Result<BoxedPipelineHandle, IndexingError>;
 
     async fn handle(
         &mut self,

quickwit/quickwit-indexing/src/actors/metrics_pipeline/indexing_service_impl.rs

@@ -19,7 +19,7 @@ use quickwit_proto::indexing::{IndexingError, IndexingPipelineId};
 
 use crate::actors::pipeline_shared::ActorPipeline;
 use crate::actors::{MetricsPipeline, MetricsPipelineParams};
-use crate::{BoxPipelineHandle, IndexingService};
+use crate::{BoxedPipelineHandle, IndexingService};
 
 impl IndexingService {
     pub(crate) async fn spawn_metrics_pipeline(
@@ -29,7 +29,7 @@ impl IndexingService {
         index_config: IndexConfig,
         source_config: SourceConfig,
         params_fingerprint: u64,
-    ) -> Result<BoxPipelineHandle, IndexingError> {
+    ) -> Result<BoxedPipelineHandle, IndexingError> {
         let pipeline_uid_str = indexing_pipeline_id.pipeline_uid.to_string();
         let indexing_directory = temp_dir::Builder::default()
             .join(&indexing_pipeline_id.index_uid.index_id)

quickwit/quickwit-indexing/src/actors/metrics_pipeline/publisher_impl.rs

@@ -72,14 +72,14 @@ impl Handler<ParquetSplitsUpdate> for Publisher {
         info!("publish-metrics-splits");
         suggest_truncate(ctx, &self.source_mailbox_opt, checkpoint_delta_opt).await;
 
-        if !split_ids.is_empty() {
+        if split_ids.is_empty() {
+            self.counters.num_empty_splits += 1;
+        } else {
             if replaced_split_ids.is_empty() {
                 self.counters.num_published_splits += 1;
             } else {
                 self.counters.num_replace_operations += 1;
             }
-        } else {
-            self.counters.num_empty_splits += 1;
         }
         Ok(())
     }

quickwit/quickwit-indexing/src/actors/mod.rs

@@ -39,7 +39,7 @@ pub use index_serializer::IndexSerializer;
 pub use indexer::{Indexer, IndexerCounters};
 pub use indexing_pipeline::{IndexingPipeline, IndexingPipelineParams};
 pub use indexing_service::{
-    BoxPipelineHandle, INDEXING_DIR_NAME, IndexingService, IndexingServiceCounters,
+    BoxedPipelineHandle, INDEXING_DIR_NAME, IndexingService, IndexingServiceCounters,
 };
 pub(crate) use log_publisher_impl::{MERGE_PUBLISHER_NAME, PUBLISHER_NAME};
 pub use merge_executor::{MergeExecutor, combine_partition_ids, merge_split_attrs};

quickwit/quickwit-indexing/src/lib.rs

@@ -26,7 +26,7 @@ use tracing::info;
 
 use crate::actors::MergeSchedulerService;
 pub use crate::actors::{
-    BoxPipelineHandle, FinishPendingMergesAndShutdownPipeline, IndexingError, IndexingPipeline,
+    BoxedPipelineHandle, FinishPendingMergesAndShutdownPipeline, IndexingError, IndexingPipeline,
     IndexingPipelineParams, IndexingService, Sequencer, SplitsUpdateMailbox,
 };
 pub use crate::controlled_directory::ControlledDirectory;