Create compaction planner service stub; wire up serving layer

quickwit/Cargo.toml

@@ -12,6 +12,7 @@ members = [
   "quickwit-control-plane",
   "quickwit-datetime",
   "quickwit-directories",
+  "quickwit-compaction",
   "quickwit-doc-mapper",
   "quickwit-index-management",
   "quickwit-indexing",
@@ -52,6 +53,7 @@ default-members = [
   "quickwit-control-plane",
   "quickwit-datetime",
   "quickwit-directories",
+  "quickwit-compaction",
   "quickwit-doc-mapper",
   "quickwit-index-management",
   "quickwit-indexing",
@@ -352,6 +354,7 @@ quickwit-control-plane = { path = "quickwit-control-plane" }
 quickwit-datetime = { path = "quickwit-datetime" }
 quickwit-directories = { path = "quickwit-directories" }
 quickwit-doc-mapper = { path = "quickwit-doc-mapper" }
+quickwit-compaction = { path = "quickwit-compaction" }
 quickwit-index-management = { path = "quickwit-index-management" }
 quickwit-indexing = { path = "quickwit-indexing" }
 quickwit-ingest = { path = "quickwit-ingest" }

quickwit/quickwit-compaction/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "quickwit-compaction"
+description = "Merge planner and merge worker services for split compaction"
+
+version.workspace = true
+edition.workspace = true
+homepage.workspace = true
+documentation.workspace = true
+repository.workspace = true
+authors.workspace = true
+license.workspace = true
+
+[dependencies]
+async-trait = { workspace = true }
+quickwit-proto = { workspace = true }

quickwit/quickwit-compaction/src/lib.rs

@@ -0,0 +1,17 @@
+// Copyright 2021-Present Datadog, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#![deny(clippy::disallowed_methods)]
+
+pub mod planner;

quickwit/quickwit-compaction/src/planner/compaction_service.rs

@@ -0,0 +1,26 @@
+// Copyright 2021-Present Datadog, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use async_trait::async_trait;
+use quickwit_proto::compaction::{CompactionResult, CompactionService, PingRequest, PingResponse};
+
+#[derive(Debug, Clone)]
+pub struct StubCompactionService;
+
+#[async_trait]
+impl CompactionService for StubCompactionService {
+    async fn ping(&self, _request: PingRequest) -> CompactionResult<PingResponse> {
+        Ok(PingResponse {})
+    }
+}

quickwit/quickwit-compaction/src/planner/mod.rs

@@ -0,0 +1,17 @@
+// Copyright 2021-Present Datadog, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+mod compaction_service;
+
+pub use compaction_service::StubCompactionService;

quickwit/quickwit-config/src/service.rs

@@ -29,6 +29,7 @@ pub enum QuickwitService {
     Searcher,
     Janitor,
     Metastore,
+    Compactor,
 }
 
 #[allow(clippy::from_over_into)]
@@ -46,6 +47,7 @@ impl QuickwitService {
             QuickwitService::Searcher => "searcher",
             QuickwitService::Janitor => "janitor",
             QuickwitService::Metastore => "metastore",
+            QuickwitService::Compactor => "compactor",
         }
     }
 
@@ -70,6 +72,7 @@ impl FromStr for QuickwitService {
             "searcher" => Ok(QuickwitService::Searcher),
             "janitor" => Ok(QuickwitService::Janitor),
             "metastore" => Ok(QuickwitService::Metastore),
+            "compactor" => Ok(QuickwitService::Compactor),
             _ => {
                 bail!(
                     "failed to parse service `{service_str}`. supported services are: `{}`",

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

@@ -1702,6 +1702,86 @@ mod tests {
         universe.quit().await;
     }
 
+    #[tokio::test]
+    async fn test_indexing_service_spawns_merge_pipeline_with_merge_scheduler() {
+        quickwit_common::setup_logging_for_tests();
+        let transport = ChannelTransport::default();
+        let cluster = create_cluster_for_test(Vec::new(), &["indexer"], &transport, true)
+            .await
+            .unwrap();
+        let metastore = metastore_for_test();
+
+        let index_id = append_random_suffix("test-indexing-service-with-merge");
+        let index_uri = format!("ram:///indexes/{index_id}");
+        let index_config = IndexConfig::for_test(&index_id, &index_uri);
+
+        let source_config = SourceConfig {
+            source_id: "test-source".to_string(),
+            num_pipelines: NonZeroUsize::MIN,
+            enabled: true,
+            source_params: SourceParams::void(),
+            transform_config: None,
+            input_format: SourceInputFormat::Json,
+        };
+        let create_index_request =
+            CreateIndexRequest::try_from_index_config(&index_config).unwrap();
+        let index_uid: IndexUid = metastore
+            .create_index(create_index_request)
+            .await
+            .unwrap()
+            .index_uid()
+            .clone();
+        let add_source_request =
+            AddSourceRequest::try_from_source_config(index_uid.clone(), &source_config).unwrap();
+        metastore.add_source(add_source_request).await.unwrap();
+
+        let temp_dir = tempfile::tempdir().unwrap();
+        let data_dir_path = temp_dir.path().to_path_buf();
+        let indexer_config = IndexerConfig::for_test().unwrap();
+        let num_blocking_threads = 1;
+        let storage_resolver = StorageResolver::unconfigured();
+        let universe = Universe::with_accelerated_time();
+        let queues_dir_path = data_dir_path.join(QUEUES_DIR_NAME);
+        let ingest_api_service =
+            init_ingest_api(&universe, &queues_dir_path, &IngestApiConfig::default())
+                .await
+                .unwrap();
+        let merge_scheduler_mailbox: Mailbox<MergeSchedulerService> = universe.get_or_spawn_one();
+        let indexing_server = IndexingService::new(
+            NodeId::from("test-node"),
+            data_dir_path,
+            indexer_config,
+            num_blocking_threads,
+            cluster.clone(),
+            metastore.clone(),
+            Some(ingest_api_service),
+            Some(merge_scheduler_mailbox),
+            IngesterPool::default(),
+            storage_resolver.clone(),
+            EventBroker::default(),
+        )
+        .await
+        .unwrap();
+        let (indexing_server_mailbox, indexing_server_handle) =
+            universe.spawn_builder().spawn(indexing_server);
+
+        indexing_server_mailbox
+            .ask_for_res(SpawnPipeline {
+                index_id: index_id.clone(),
+                source_config,
+                pipeline_uid: PipelineUid::default(),
+            })
+            .await
+            .unwrap();
+
+        let observation = indexing_server_handle.observe().await;
+        assert_eq!(observation.num_running_pipelines, 1);
+        assert_eq!(observation.num_running_merge_pipelines, 1);
+        assert!(universe.get_one::<MergePipeline>().is_some());
+
+        universe.quit().await;
+    }
+
     #[derive(Debug)]
     struct FreezePipeline;
     #[async_trait]

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

@@ -213,16 +213,12 @@ impl Handler<SplitsUpdate> for Publisher<DocProcessor> {
             // When merging is handled locally, notify the merge planner about new
             // splits. The mailbox is None when an external merge service is active,
             // or when the planner has already shut down (e.g. source reached its end).
-            if let Some(merge_planner_mailbox) = &self.merge_planner_mailbox_opt {
-                match ctx
+            if let Some(merge_planner_mailbox) = &self.merge_planner_mailbox_opt
+                && let Err(error) = ctx
                     .send_message(merge_planner_mailbox, NewSplits { new_splits })
                     .await
-                {
-                    Ok(_) => {}
-                    Err(error) => {
-                        error!(error=?error, "failed to send new splits to merge planner");
-                    }
-                }
+            {
+                error!(error=?error, "failed to send new splits to merge planner");
             }
 
             if replaced_split_ids.is_empty() {

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

@@ -162,3 +162,99 @@ async fn test_multi_nodes_cluster() {
 
     sandbox.shutdown().await.unwrap();
 }
+
+#[tokio::test]
+async fn test_no_merge_pipelines_when_compaction_service_enabled() {
+    quickwit_common::setup_logging_for_tests();
+    unsafe { std::env::set_var("QW_ENABLE_COMPACTION_SERVICE", "true") };
+
+    let sandbox = ClusterSandboxBuilder::default()
+        .add_node([QuickwitService::Searcher])
+        .add_node([QuickwitService::Metastore])
+        .add_node([QuickwitService::Indexer])
+        .add_node([QuickwitService::ControlPlane])
+        .add_node([QuickwitService::Janitor])
+        .build_and_start()
+        .await;
+
+    sandbox
+        .rest_client(QuickwitService::Indexer)
+        .indexes()
+        .create(
+            r#"
+            version: 0.8
+            index_id: test-no-merge-pipelines
+            doc_mapping:
+              field_mappings:
+              - name: body
+                type: text
+            indexing_settings:
+              commit_timeout_secs: 1
+            "#,
+            quickwit_config::ConfigFormat::Yaml,
+            false,
+        )
+        .await
+        .unwrap();
+
+    sandbox.wait_for_indexing_pipelines(1).await.unwrap();
+
+    let stats = sandbox
+        .rest_client(QuickwitService::Indexer)
+        .node_stats()
+        .indexing()
+        .await
+        .unwrap();
+    assert_eq!(stats.num_running_merge_pipelines, 0);
+
+    unsafe { std::env::remove_var("QW_ENABLE_COMPACTION_SERVICE") };
+    sandbox.shutdown().await.unwrap();
+}
+
+#[tokio::test]
+async fn test_merge_pipelines_present_without_compaction_service() {
+    quickwit_common::setup_logging_for_tests();
+    unsafe { std::env::set_var("QW_ENABLE_COMPACTION_SERVICE", "false") };
+
+    let sandbox = ClusterSandboxBuilder::default()
+        .add_node([QuickwitService::Searcher])
+        .add_node([QuickwitService::Metastore])
+        .add_node([QuickwitService::Indexer])
+        .add_node([QuickwitService::ControlPlane])
+        .add_node([QuickwitService::Janitor])
+        .build_and_start()
+        .await;
+
+    sandbox
+        .rest_client(QuickwitService::Indexer)
+        .indexes()
+        .create(
+            r#"
+            version: 0.8
+            index_id: test-with-merge-pipelines
+            doc_mapping:
+              field_mappings:
+              - name: body
+                type: text
+            indexing_settings:
+              commit_timeout_secs: 1
+            "#,
+            quickwit_config::ConfigFormat::Yaml,
+            false,
+        )
+        .await
+        .unwrap();
+
+    sandbox.wait_for_indexing_pipelines(1).await.unwrap();
+
+    let stats = sandbox
+        .rest_client(QuickwitService::Indexer)
+        .node_stats()
+        .indexing()
+        .await
+        .unwrap();
+    assert!(stats.num_running_merge_pipelines > 0);
+
+    unsafe { std::env::remove_var("QW_ENABLE_COMPACTION_SERVICE") };
+    sandbox.shutdown().await.unwrap();
+}

quickwit/quickwit-proto/build.rs

@@ -35,6 +35,20 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         .run()
         .unwrap();
 
+    // Compaction service.
+    let mut prost_config = prost_build::Config::default();
+    prost_config.file_descriptor_set_path("src/codegen/quickwit/compaction_descriptor.bin");
+
+    Codegen::builder()
+        .with_prost_config(prost_config)
+        .with_protos(&["protos/quickwit/compaction.proto"])
+        .with_output_dir("src/codegen/quickwit")
+        .with_result_type_path("crate::compaction::CompactionResult")
+        .with_error_type_path("crate::compaction::CompactionError")
+        .generate_rpc_name_impls()
+        .run()
+        .unwrap();
+
     // Control plane.
     let mut prost_config = prost_build::Config::default();
     prost_config.file_descriptor_set_path("src/codegen/quickwit/control_plane_descriptor.bin");