fix: thread ParquetWriterConfig through pipeline, fix misleading retry comment

Review findings:

1. ParquetWriterConfig was hardcoded to Default in the executor. If ingest
   uses custom compression, merge output would differ. Now threaded from
   ParquetMergePipelineParams through to the executor.

2. Fixed misleading comment claiming "planner will eventually re-plan"
   on merge failure. In reality, input splits are drained by operations()
   and won't be re-planned until the pipeline restarts with metastore
   re-seeding (not yet implemented — TODO added).

3. Added TODO for fetch_immature_parquet_splits() on pipeline respawn,
   matching the Tantivy MergePipeline pattern.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: g-talbot

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

@@ -43,11 +43,18 @@ use crate::models::PublishLock;
 /// ready-to-upload Parquet files with complete metadata.
 pub struct ParquetMergeExecutor {
     uploader_mailbox: Mailbox<ParquetUploader>,
+    writer_config: ParquetWriterConfig,
 }
 
 impl ParquetMergeExecutor {
-    pub fn new(uploader_mailbox: Mailbox<ParquetUploader>) -> Self {
-        Self { uploader_mailbox }
+    pub fn new(
+        uploader_mailbox: Mailbox<ParquetUploader>,
+        writer_config: ParquetWriterConfig,
+    ) -> Self {
+        Self {
+            uploader_mailbox,
+            writer_config,
+        }
     }
 }
 
@@ -98,36 +105,38 @@ impl Handler<ParquetMergeScratch> for ParquetMergeExecutor {
         // Run the CPU-intensive merge on the dedicated thread pool.
         let input_paths = scratch.downloaded_parquet_files.clone();
         let output_dir_clone = output_dir.clone();
+        let writer_config = self.writer_config.clone();
         let merge_result = run_cpu_intensive(move || {
             let config = MergeConfig {
                 num_outputs: 1,
-                writer_config: ParquetWriterConfig::default(),
+                writer_config,
             };
             merge_sorted_parquet_files(&input_paths, &output_dir_clone, &config)
         })
         .await;
 
-        // We return Ok(()) on merge failure rather than Err to keep the actor
-        // alive — same strategy as Tantivy's MergeExecutor. This prevents a
-        // single "split of death" from crash-looping the entire pipeline.
-        // The trade-off: failed splits aren't retried until pipeline respawn.
         let outputs: Vec<MergeOutputFile> = match merge_result {
             Ok(Ok(outputs)) => outputs,
             Ok(Err(merge_err)) => {
                 warn!(
                     error = %merge_err,
                     merge_split_id = %merge_split_id,
-                    "parquet merge failed"
+                    "parquet merge failed — input splits will not be retried until \
+                     the pipeline restarts with metastore re-seeding"
                 );
-                // Input splits were drained from the planner by operations().
-                // They remain published but won't be re-planned until respawn.
+                // The input splits were drained from the planner by operations().
+                // They remain published in the metastore but won't be re-planned
+                // until the pipeline restarts and re-seeds from the metastore.
+                // TODO: implement fetch_immature_parquet_splits() for respawn
+                // (same as Tantivy's fetch_immature_splits pattern).
                 return Ok(());
             }
             Err(panicked) => {
                 warn!(
                     error = %panicked,
                     merge_split_id = %merge_split_id,
-                    "parquet merge panicked"
+                    "parquet merge panicked — input splits will not be retried until \
+                     the pipeline restarts with metastore re-seeding"
                 );
                 return Ok(());
             }
@@ -160,9 +169,8 @@ impl Handler<ParquetMergeScratch> for ParquetMergeExecutor {
                 .context("failed to build merge output metadata")
                 .map_err(|e| ActorExitStatus::from(anyhow::anyhow!(e)))?;
 
-            // The merge engine writes to a temp filename (merge_output_*.parquet).
-            // Rename to {split_id}.parquet so the uploader can find it at the
-            // path derived from ParquetSplitMetadata::parquet_filename().
+            // Rename the output file to match the generated split ID.
+            // The uploader expects files at `output_dir/{split_id}.parquet`.
             let expected_path = output_dir.join(&metadata.parquet_file);
             if output.path != expected_path {
                 std::fs::rename(&output.path, &expected_path)

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

@@ -225,6 +225,11 @@ impl ParquetMergePipeline {
             "spawning parquet merge pipeline"
         );
 
+        // On first spawn, use the initial splits provided by the IndexingService.
+        // On subsequent spawns (after crash/respawn), we start empty.
+        // TODO: implement fetch_immature_parquet_splits() to re-query the
+        // metastore on respawn, matching the Tantivy MergePipeline pattern
+        // (see merge_pipeline.rs:441-471).
         let immature_splits = self.initial_immature_splits_opt.take().unwrap_or_default();
 
         // Spawn actors bottom-up: each actor's constructor needs a mailbox
@@ -269,7 +274,8 @@ impl ParquetMergePipeline {
             .spawn(merge_uploader);
 
         // 4. Merge executor
-        let merge_executor = ParquetMergeExecutor::new(merge_uploader_mailbox);
+        let merge_executor =
+            ParquetMergeExecutor::new(merge_uploader_mailbox, self.params.writer_config.clone());
         let (merge_executor_mailbox, merge_executor_handle) = ctx
             .spawn_actor()
             .set_kill_switch(self.kill_switch.clone())
@@ -447,6 +453,10 @@ pub struct ParquetMergePipelineParams {
     pub merge_scheduler_service: Mailbox<MergeSchedulerService>,
     pub max_concurrent_split_uploads: usize,
     pub event_broker: EventBroker,
+    /// Parquet writer config for merge output (compression, page size, etc.).
+    /// Should match the ingest pipeline's writer config so merged files have
+    /// consistent compression.
+    pub writer_config: quickwit_parquet_engine::storage::ParquetWriterConfig,
 }
 
 #[cfg(test)]
@@ -484,6 +494,7 @@ mod tests {
             merge_scheduler_service: universe.get_or_spawn_one(),
             max_concurrent_split_uploads: 4,
             event_broker: EventBroker::default(),
+            writer_config: quickwit_parquet_engine::storage::ParquetWriterConfig::default(),
         }
     }
 

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

@@ -178,6 +178,7 @@ async fn test_merge_pipeline_end_to_end() {
         merge_scheduler_service: universe.get_or_spawn_one(),
         max_concurrent_split_uploads: 4,
         event_broker: EventBroker::default(),
+        writer_config: ParquetWriterConfig::default(),
     };
 
     let initial_splits = vec![meta_a, meta_b];