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,10 +105,11 @@ 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)
         })
@@ -113,16 +121,22 @@ impl Handler<ParquetMergeScratch> for ParquetMergeExecutor {
                 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"
                 );
-                // Drop the merge permit — the planner will eventually re-plan.
+                // 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(());
             }

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 so downstream mailboxes are available.
@@ -265,7 +270,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())
@@ -443,6 +449,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)]
@@ -480,6 +490,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];