feat: wire ParquetIndexingConfig and ParquetMergePolicyConfig to pipeline

Reads parquet_indexing.sort_fields and parquet_indexing.window_duration_secs
from IndexingSettings when constructing the ingest pipeline's TableConfig
(was hardcoded to defaults).

Adds parquet_merge_policy_from_settings() that converts the config-layer
ParquetMergePolicyConfig to an Arc<dyn ParquetMergePolicy> runtime policy,
paralleling merge_policy_from_settings() for Tantivy.

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

Author: g-talbot

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

@@ -346,9 +346,14 @@ impl MetricsPipeline {
             .set_kill_switch(self.kill_switch.clone())
             .spawn(uploader);
 
-        // ParquetPackager
+        // ParquetPackager — read sort schema and window duration from index config.
         let writer_config = quickwit_parquet_engine::storage::ParquetWriterConfig::default();
-        let table_config = quickwit_parquet_engine::table_config::TableConfig::default();
+        let parquet_indexing_config = &self.params.indexing_settings.parquet_indexing;
+        let mut table_config = quickwit_parquet_engine::table_config::TableConfig::default();
+        if let Some(ref sort_fields) = parquet_indexing_config.sort_fields {
+            table_config.sort_fields = Some(sort_fields.clone());
+        }
+        table_config.window_duration_secs = parquet_indexing_config.window_duration_secs;
         let split_kind = if self.params.use_sketch_processors {
             quickwit_parquet_engine::split::ParquetSplitKind::Sketches
         } else {

quickwit/quickwit-indexing/src/merge_policy/mod.rs

@@ -192,6 +192,27 @@ pub fn default_merge_policy() -> Arc<dyn MergePolicy> {
     merge_policy_from_settings(&indexing_settings)
 }
 
+/// Creates a Parquet merge policy from the index's `ParquetMergePolicyConfig`.
+#[cfg(feature = "metrics")]
+pub fn parquet_merge_policy_from_settings(
+    settings: &IndexingSettings,
+) -> Arc<dyn quickwit_parquet_engine::merge::policy::ParquetMergePolicy> {
+    let config = &settings.parquet_merge_policy;
+    let engine_config = quickwit_parquet_engine::merge::policy::ParquetMergePolicyConfig {
+        merge_factor: config.merge_factor,
+        max_merge_factor: config.max_merge_factor,
+        max_merge_ops: config.max_merge_ops,
+        target_split_size_bytes: config.target_split_size_bytes,
+        maturation_period: config.maturation_period,
+        max_finalize_merge_operations: config.max_finalize_merge_operations,
+    };
+    Arc::new(
+        quickwit_parquet_engine::merge::policy::ConstWriteAmplificationParquetMergePolicy::new(
+            engine_config,
+        ),
+    )
+}
+
 pub fn nop_merge_policy() -> Arc<dyn MergePolicy> {
     Arc::new(NopMergePolicy)
 }