feat: Phase 3d+3e — ParquetMergePipeline supervisor + publisher feedback (#6354)

* feat: add ParquetMergePipeline supervisor and publisher feedback (Phase 3d+3e)

Phase 3 pipeline integration, combined supervisor and feedback PR:

- ParquetMergePipeline supervisor: spawns all merge actors (publisher →
  sequencer → uploader → executor → downloader → planner), health-checks
  with periodic supervision loop, respawn on failure with backoff,
  graceful shutdown via FinishPendingMergesAndShutdownPipeline that
  disconnects feedback and runs finalize policy. 3 tests.

- Publisher feedback: add parquet_merge_planner_mailbox_opt to Publisher
  (feature-gated behind cfg(feature = "metrics")). After successful
  ParquetSplitsUpdate publish of new ingested splits, sends ParquetNewSplits
  to the planner. Merge outputs (non-empty replaced_split_ids) are not
  fed back to avoid infinite loops.

- DisconnectMergePlanner extended to clear both Tantivy and Parquet planner
  mailboxes, supporting shutdown drain for both pipeline types.

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

* fix: nightly rustfmt import ordering in parquet_merge_pipeline

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

* fix: reflow comment to satisfy nightly rustfmt wrap_comments

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

* fix: feed back merge outputs to planner, matching Tantivy behavior

The guard that filtered out merge outputs (non-empty replaced_split_ids)
from the feedback loop was incorrect. Tantivy feeds ALL new splits to
the merge planner — both ingest and merge outputs. Infinite loops are
prevented by the merge policy's maturity checks (max_merge_ops, size,
maturation_period), not by the publisher.

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

---------

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

Author: g-talbot

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

@@ -39,8 +39,16 @@ impl Handler<DisconnectMergePlanner> for Publisher {
         _: DisconnectMergePlanner,
         _ctx: &ActorContext<Self>,
     ) -> Result<(), ActorExitStatus> {
+        // Clear both Tantivy and Parquet planner mailboxes. Each Publisher
+        // instance only has one of these set (depending on which pipeline it
+        // serves), but clearing both is safe and avoids needing separate
+        // disconnect message types.
         info!("disconnecting merge planner mailbox");
         self.merge_planner_mailbox_opt = None;
+        #[cfg(feature = "metrics")]
+        {
+            self.parquet_merge_planner_mailbox_opt = None;
+        }
         Ok(())
     }
 }

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

@@ -27,6 +27,7 @@ mod parquet_doc_processor;
 mod parquet_indexer;
 mod parquet_merge_executor;
 pub(crate) mod parquet_merge_messages;
+mod parquet_merge_pipeline;
 mod parquet_merge_planner;
 mod parquet_merge_split_downloader;
 mod parquet_packager;
@@ -50,6 +51,7 @@ pub use parquet_doc_processor::{
 pub use parquet_indexer::{ParquetIndexer, ParquetIndexerCounters, ParquetSplitBatch};
 pub use parquet_merge_executor::ParquetMergeExecutor;
 pub use parquet_merge_messages::{ParquetMergeScratch, ParquetMergeTask, ParquetNewSplits};
+pub use parquet_merge_pipeline::{ParquetMergePipeline, ParquetMergePipelineParams};
 pub use parquet_merge_planner::ParquetMergePlanner;
 pub use parquet_merge_split_downloader::ParquetMergeSplitDownloader;
 pub use parquet_packager::{ParquetBatchForPackager, ParquetPackager, ParquetPackagerCounters};