[metrics] Support DDSketch in the parquet pipeline (#6257)

* generalize metric splits to parquet splits

* add sketch schema + processor

* sketch split support in metastore

* wire sketch pipeline into indexing actors and control plane

* drive by fix: pass sort fields to bloom filter config

* address comments, lint

* address comments, lint

* linter

* linter

* fix unit test

Author: mattmkim

quickwit/Cargo.lock

@@ -20,9 +20,22 @@ pub fn is_metrics_index(index_id: &str) -> bool {
     index_id.starts_with("otel-metrics") || index_id.starts_with("metrics-")
 }
 
+/// Returns whether the given index ID corresponds to a sketches index.
+///
+/// Sketches indexes use the Parquet/DataFusion pipeline with sketch-specific
+/// processors and writers.
+pub fn is_sketches_index(index_id: &str) -> bool {
+    index_id.starts_with("sketches-")
+}
+
+/// Returns whether the given index ID uses the Parquet/DataFusion pipeline.
+pub fn is_parquet_pipeline_index(index_id: &str) -> bool {
+    is_metrics_index(index_id) || is_sketches_index(index_id)
+}
+
 #[cfg(test)]
 mod tests {
-    use super::is_metrics_index;
+    use super::*;
 
     #[test]
     fn test_is_metrics_index() {
@@ -44,4 +57,19 @@ mod tests {
         assert!(!is_metrics_index("metrics")); // No hyphen after "metrics"
         assert!(!is_metrics_index("my-metrics-index")); // Not prefixed
     }
+
+    #[test]
+    fn test_is_sketches_index() {
+        assert!(is_sketches_index("sketches-default"));
+        assert!(!is_sketches_index("otel-metrics"));
+        assert!(!is_sketches_index("my-index"));
+    }
+
+    #[test]
+    fn test_is_parquet_pipeline_index() {
+        assert!(is_parquet_pipeline_index("otel-metrics"));
+        assert!(is_parquet_pipeline_index("sketches-default"));
+        assert!(!is_parquet_pipeline_index("otel-logs-v0_7"));
+        assert!(!is_parquet_pipeline_index("my-index"));
+    }
 }

quickwit/quickwit-common/src/metrics_specific.rs

@@ -23,7 +23,7 @@ use std::time::{Duration, Instant};
 
 use fnv::{FnvHashMap, FnvHashSet};
 use itertools::Itertools;
-use quickwit_common::is_metrics_index;
+use quickwit_common::is_parquet_pipeline_index;
 use quickwit_common::pretty::PrettySample;
 use quickwit_config::{FileSourceParams, SourceParams, indexing_pipeline_params_fingerprint};
 use quickwit_proto::indexing::{
@@ -217,7 +217,7 @@ fn get_sources_to_schedule(model: &ControlPlaneModel) -> Vec<SourceToSchedule> {
             SourceParams::IngestApi => {
                 // Metrics indexes should use IngestV2 only, not IngestV1.
                 // The ParquetSourceLoader doesn't support IngestV1.
-                if is_metrics_index(&source_uid.index_uid.index_id) {
+                if is_parquet_pipeline_index(&source_uid.index_uid.index_id) {
                     continue;
                 }
                 // TODO ingest v1 is scheduled differently

quickwit/quickwit-control-plane/src/indexing_scheduler/mod.rs

@@ -18,7 +18,7 @@
 // In that case, metrics index will be started as a metrics pipeline.
 
 use quickwit_actors::ActorContext;
-use quickwit_common::{is_metrics_index, temp_dir};
+use quickwit_common::{is_parquet_pipeline_index, is_sketches_index, temp_dir};
 use quickwit_config::{IndexConfig, SourceConfig};
 use quickwit_metastore::SplitMetadata;
 use quickwit_proto::indexing::{IndexingError, IndexingPipelineId};
@@ -35,6 +35,7 @@ impl IndexingService {
         index_config: IndexConfig,
         source_config: SourceConfig,
         params_fingerprint: u64,
+        use_sketch_processors: bool,
     ) -> Result<BoxedPipelineHandle, IndexingError> {
         let pipeline_uid_str = indexing_pipeline_id.pipeline_uid.to_string();
         let indexing_directory = temp_dir::Builder::default()
@@ -69,6 +70,7 @@ impl IndexingService {
             source_storage_resolver: self.storage_resolver.clone(),
             params_fingerprint,
             event_broker: self.event_broker.clone(),
+            use_sketch_processors,
         };
         let pipeline = MetricsPipeline::new(pipeline_params);
         let (mailbox, handle) = ctx.spawn_actor().spawn(pipeline);
@@ -88,13 +90,16 @@ impl IndexingService {
         immature_splits_opt: Option<Vec<SplitMetadata>>,
         params_fingerprint: u64,
     ) -> Result<BoxedPipelineHandle, IndexingError> {
-        if is_metrics_index(&indexing_pipeline_id.index_uid.index_id) {
+        let index_id = &indexing_pipeline_id.index_uid.index_id;
+        if is_parquet_pipeline_index(index_id) {
+            let use_sketch_processors = is_sketches_index(index_id);
             self.spawn_metrics_pipeline(
                 ctx,
                 indexing_pipeline_id.clone(),
                 index_config,
                 source_config,
                 params_fingerprint,
+                use_sketch_processors,
             )
             .await
         } else {

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

@@ -23,7 +23,9 @@ use quickwit_actors::{Actor, ActorContext, ActorExitStatus, Handler, Mailbox, Qu
 use quickwit_common::rate_limited_tracing::rate_limited_warn;
 use quickwit_common::runtimes::RuntimeType;
 use quickwit_metastore::checkpoint::SourceCheckpointDelta;
-use quickwit_parquet_engine::ingest::{IngestError, ParquetIngestProcessor};
+use quickwit_parquet_engine::ingest::{
+    IngestError, ParquetIngestProcessor, SketchParquetIngestProcessor,
+};
 use quickwit_proto::types::{IndexId, SourceId};
 use serde::Serialize;
 use tokio::runtime::Handle;
@@ -114,17 +116,32 @@ pub enum ParquetDocProcessorError {
     Ingest(#[from] IngestError),
 }
 
-/// ParquetDocProcessor actor that routes Arrow IPC batches to the metrics engine.
+/// Enum wrapping the ingest processor variant for metrics vs sketches.
+pub enum IngestProcessor {
+    Metrics(ParquetIngestProcessor),
+    Sketches(SketchParquetIngestProcessor),
+}
+
+impl IngestProcessor {
+    fn process_ipc(&self, ipc_bytes: &[u8]) -> Result<RecordBatch, IngestError> {
+        match self {
+            Self::Metrics(p) => p.process_ipc(ipc_bytes),
+            Self::Sketches(p) => p.process_ipc(ipc_bytes),
+        }
+    }
+}
+
+/// ParquetDocProcessor actor that routes Arrow IPC batches to the parquet engine.
 ///
 /// This actor receives RawDocBatch messages containing Arrow IPC data and converts
-/// them to RecordBatch using ParquetIngestProcessor. The resulting batches are
-/// forwarded to ParquetIndexer for accumulation and split production.
+/// them to RecordBatch using the configured ingest processor. The resulting batches
+/// are forwarded to ParquetIndexer for accumulation and split production.
 ///
 /// Unlike DocProcessor which converts to Tantivy documents, this actor works
-/// exclusively with Arrow RecordBatch for high-throughput metrics ingestion.
+/// exclusively with Arrow RecordBatch for high-throughput metrics/sketch ingestion.
 pub struct ParquetDocProcessor {
     /// Processor for converting Arrow IPC to RecordBatch.
-    processor: ParquetIngestProcessor,
+    processor: IngestProcessor,
     /// Processing counters.
     counters: ParquetDocProcessorCounters,
     /// Publish lock for coordinating with sources.
@@ -136,11 +153,11 @@ pub struct ParquetDocProcessor {
 impl ParquetDocProcessor {
     /// Creates a new ParquetDocProcessor.
     pub fn new(
+        processor: IngestProcessor,
         index_id: IndexId,
         source_id: SourceId,
         indexer_mailbox: Mailbox<ParquetIndexer>,
     ) -> Self {
-        let processor = ParquetIngestProcessor;
         let counters = ParquetDocProcessorCounters::new(index_id.clone(), source_id.clone());
 
         info!(
@@ -401,6 +418,7 @@ mod tests {
 
         let (indexer_mailbox, _indexer_inbox) = universe.create_test_mailbox::<ParquetIndexer>();
         let metrics_doc_processor = ParquetDocProcessor::new(
+            IngestProcessor::Metrics(ParquetIngestProcessor),
             "test-metrics-index".to_string(),
             "test-source".to_string(),
             indexer_mailbox,
@@ -442,6 +460,7 @@ mod tests {
 
         let (indexer_mailbox, _indexer_inbox) = universe.create_test_mailbox::<ParquetIndexer>();
         let metrics_doc_processor = ParquetDocProcessor::new(
+            IngestProcessor::Metrics(ParquetIngestProcessor),
             "test-metrics-index".to_string(),
             "test-source".to_string(),
             indexer_mailbox,
@@ -477,6 +496,7 @@ mod tests {
 
         let (indexer_mailbox, _indexer_inbox) = universe.create_test_mailbox::<ParquetIndexer>();
         let metrics_doc_processor = ParquetDocProcessor::new(
+            IngestProcessor::Metrics(ParquetIngestProcessor),
             "test-metrics-index".to_string(),
             "test-source".to_string(),
             indexer_mailbox,
@@ -547,7 +567,12 @@ mod tests {
         // Create ParquetPackager
         let writer_config = ParquetWriterConfig::default();
         let table_config = quickwit_parquet_engine::table_config::TableConfig::default();
-        let split_writer = ParquetSplitWriter::new(writer_config, temp_dir.path(), &table_config);
+        let split_writer = ParquetSplitWriter::new(
+            quickwit_parquet_engine::split::ParquetSplitKind::Metrics,
+            writer_config,
+            temp_dir.path(),
+            &table_config,
+        );
         let packager = ParquetPackager::new(split_writer, uploader_mailbox);
         let (packager_mailbox, packager_handle) = universe.spawn_builder().spawn(packager);
 
@@ -562,6 +587,7 @@ mod tests {
         let (indexer_mailbox, indexer_handle) = universe.spawn_builder().spawn(indexer);
 
         let metrics_doc_processor = ParquetDocProcessor::new(
+            IngestProcessor::Metrics(ParquetIngestProcessor),
             "test-index".to_string(),
             "test-source".to_string(),
             indexer_mailbox,