feat: add merge metadata aggregation, message types, and replaced_split_ids (Phase 3a)

Phase 3 pipeline integration, first PR:

- merge_parquet_split_metadata(): aggregates input split metadata with
  MergeOutputFile physical metadata to produce complete ParquetSplitMetadata
  for merged output. Validates invariant fields, unions metric_names and tags,
  finalizes tag cardinality after merge. 17 tests.

- ParquetNewSplits, ParquetMergeTask, ParquetMergeScratch message types for
  the merge actor chain (planner → scheduler → downloader → executor).

- Add replaced_split_ids to ParquetSplitBatch and propagate through
  ParquetUploader (was hardcoded Vec::new()). Enables merge executor to
  specify which splits are being replaced.

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

Author: g-talbot

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

@@ -25,6 +25,7 @@
 mod indexing_service_impl;
 mod parquet_doc_processor;
 mod parquet_indexer;
+pub(crate) mod parquet_merge_messages;
 mod parquet_packager;
 mod parquet_splits_update;
 mod parquet_uploader;
@@ -44,6 +45,7 @@ pub use parquet_doc_processor::{
     ParquetDocProcessor, ParquetDocProcessorCounters, ParquetDocProcessorError, is_arrow_ipc,
 };
 pub use parquet_indexer::{ParquetIndexer, ParquetIndexerCounters, ParquetSplitBatch};
+pub use parquet_merge_messages::{ParquetMergeScratch, ParquetMergeTask, ParquetNewSplits};
 pub use parquet_packager::{ParquetBatchForPackager, ParquetPackager, ParquetPackagerCounters};
 pub use parquet_splits_update::ParquetSplitsUpdate;
 pub use parquet_uploader::ParquetUploader;

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

@@ -100,6 +100,9 @@ pub struct ParquetSplitBatch {
     pub publish_lock: PublishLock,
     /// Optional publish token.
     pub publish_token_opt: Option<PublishToken>,
+    /// Split IDs being replaced by this batch (non-empty for merges).
+    /// Empty for the ingest path.
+    pub replaced_split_ids: Vec<String>,
 }
 
 /// ParquetIndexer actor that accumulates RecordBatches and forwards them to ParquetPackager.

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

@@ -0,0 +1,91 @@
+// 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.
+
+//! Message types for the Parquet merge pipeline.
+//!
+//! These messages flow through the actor chain:
+//!
+//! ```text
+//! ParquetMergePlanner ──► MergeSchedulerService ──► ParquetMergeSplitDownloader
+//!                                                         │
+//!                                                         ▼ (ParquetMergeScratch)
+//!                                                   ParquetMergeExecutor
+//! ```
+
+use std::fmt;
+use std::path::PathBuf;
+
+use quickwit_common::temp_dir::TempDirectory;
+use quickwit_parquet_engine::merge::policy::ParquetMergeOperation;
+use quickwit_parquet_engine::split::ParquetSplitMetadata;
+use tantivy::TrackedObject;
+
+use crate::actors::MergePermit;
+
+/// Notification of newly created Parquet splits.
+///
+/// Sent to `ParquetMergePlanner` from:
+/// - The publisher (feedback loop after a merge completes)
+/// - The indexing service (initial seeding of immature splits on start)
+#[derive(Debug)]
+pub struct ParquetNewSplits {
+    pub new_splits: Vec<ParquetSplitMetadata>,
+}
+
+/// A merge task dispatched by `MergeSchedulerService` to `ParquetMergeSplitDownloader`.
+///
+/// Carries the merge operation (tracked by the planner's inventory) and a
+/// concurrency permit from the global merge semaphore.
+pub struct ParquetMergeTask {
+    pub merge_operation: TrackedObject<ParquetMergeOperation>,
+    pub merge_permit: MergePermit,
+}
+
+impl fmt::Debug for ParquetMergeTask {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("ParquetMergeTask")
+            .field("merge_split_id", &self.merge_operation.merge_split_id)
+            .field("num_inputs", &self.merge_operation.splits.len())
+            .finish()
+    }
+}
+
+/// Downloaded Parquet files ready for merge execution.
+///
+/// Sent from `ParquetMergeSplitDownloader` to `ParquetMergeExecutor` after
+/// all input files have been downloaded to local storage.
+pub struct ParquetMergeScratch {
+    /// The merge operation describing what to merge.
+    pub merge_operation: TrackedObject<ParquetMergeOperation>,
+
+    /// Local paths to the downloaded Parquet files, one per input split,
+    /// in the same order as `merge_operation.splits`.
+    pub downloaded_parquet_files: Vec<PathBuf>,
+
+    /// Temp directory containing the downloaded files. Held to prevent cleanup
+    /// until the merge executor is done with the files.
+    pub scratch_directory: TempDirectory,
+
+    /// Concurrency permit — held until the merge completes (including upload).
+    pub merge_permit: MergePermit,
+}
+
+impl fmt::Debug for ParquetMergeScratch {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("ParquetMergeScratch")
+            .field("merge_split_id", &self.merge_operation.merge_split_id)
+            .field("num_files", &self.downloaded_parquet_files.len())
+            .finish()
+    }
+}

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

@@ -236,6 +236,7 @@ impl Handler<ParquetBatchForPackager> for ParquetPackager {
             checkpoint_delta,
             publish_lock,
             publish_token_opt,
+            replaced_split_ids: Vec::new(),
         };
 
         ctx.send_message(&self.uploader_mailbox, split_batch)

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

@@ -221,6 +221,7 @@ impl Handler<ParquetSplitBatch> for ParquetUploader {
         let publish_lock = batch.publish_lock;
         let publish_token_opt = batch.publish_token_opt;
         let splits = batch.splits;
+        let replaced_split_ids = batch.replaced_split_ids;
         debug!(
             index_uid = %index_uid,
             num_splits = splits.len(),
@@ -321,7 +322,7 @@ impl Handler<ParquetSplitBatch> for ParquetUploader {
                 let update = ParquetSplitsUpdate {
                     index_uid,
                     new_splits: splits,
-                    replaced_split_ids: Vec::new(), // No merging yet
+                    replaced_split_ids,
                     checkpoint_delta_opt: Some(checkpoint_delta),
                     publish_lock,
                     publish_token_opt,
@@ -427,6 +428,7 @@ mod tests {
             checkpoint_delta,
             publish_lock: PublishLock::default(),
             publish_token_opt: None,
+            replaced_split_ids: Vec::new(),
         };
 
         uploader_mailbox.send_message(batch).await.unwrap();
@@ -520,6 +522,7 @@ mod tests {
             checkpoint_delta,
             publish_lock: PublishLock::default(),
             publish_token_opt: None,
+            replaced_split_ids: Vec::new(),
         };
 
         uploader_mailbox.send_message(batch).await.unwrap();
@@ -594,6 +597,7 @@ mod tests {
             checkpoint_delta,
             publish_lock: PublishLock::default(),
             publish_token_opt: None,
+            replaced_split_ids: Vec::new(),
         };
 
         uploader_mailbox.send_message(batch).await.unwrap();
@@ -664,6 +668,7 @@ mod tests {
                 checkpoint_delta,
                 publish_lock: PublishLock::default(),
                 publish_token_opt: None,
+                replaced_split_ids: Vec::new(),
             };
             uploader_mailbox.send_message(batch).await.unwrap();
         }