trying to isolate code from metrics

quickwit/quickwit-cli/Cargo.toml

@@ -100,6 +100,7 @@ release-feature-set = [
   "pprof",
   "quickwit-indexing/kafka",
   "quickwit-indexing/kinesis",
+  "quickwit-indexing/metrics",
   "quickwit-indexing/pulsar",
   "quickwit-indexing/sqs",
   "quickwit-indexing/vrl",
@@ -113,6 +114,7 @@ release-feature-vendored-set = [
   "openssl-support",
   "pprof",
   "quickwit-indexing/kinesis",
+  "quickwit-indexing/metrics",
   "quickwit-indexing/pulsar",
   "quickwit-indexing/sqs",
   "quickwit-indexing/vrl",
@@ -126,6 +128,7 @@ release-macos-feature-vendored-set = [
   "jemalloc",
   "openssl-support",
   "quickwit-indexing/kinesis",
+  "quickwit-indexing/metrics",
   "quickwit-indexing/pulsar",
   "quickwit-indexing/sqs",
   "quickwit-indexing/vrl",

quickwit/quickwit-cli/src/tool.rs

@@ -37,7 +37,6 @@ use quickwit_config::{
     TransformConfig, VecSourceParams,
 };
 use quickwit_index_management::{IndexService, clear_cache_directory};
-use quickwit_indexing::IndexingPipeline;
 use quickwit_indexing::actors::{IndexingService, MergePipeline, MergeSchedulerService};
 use quickwit_indexing::models::{
     DetachIndexingPipeline, DetachMergePipeline, IndexingStatistics, SpawnPipeline,
@@ -751,7 +750,7 @@ async fn extract_split_cli(args: ExtractSplitArgs) -> anyhow::Result<()> {
 /// Starts a tokio task that displays the indexing statistics
 /// every once in awhile.
 pub async fn start_statistics_reporting_loop(
-    pipeline_handle: ActorHandle<IndexingPipeline>,
+    pipeline_handle: quickwit_indexing::BoxPipelineHandle,
     is_stdin: bool,
 ) -> anyhow::Result<IndexingStatistics> {
     let mut stdout_handle = stdout();

quickwit/quickwit-common/src/lib.rs

@@ -49,6 +49,9 @@ pub mod tower;
 pub mod type_map;
 pub mod uri;
 
+mod metrics_specific;
+pub use metrics_specific::*;
+
 mod socket_addr_legacy_hash;
 
 use std::env;
@@ -215,14 +218,6 @@ macro_rules! assert_eventually {
     };
 }
 
-/// Returns whether the given index ID corresponds to a metrics index.
-///
-/// Metrics indexes use the Parquet/DataFusion pipeline instead of the Tantivy pipeline.
-/// An index is considered a metrics index if it starts with "otel-metrics" or "metrics-".
-pub fn is_metrics_index(index_id: &str) -> bool {
-    index_id.starts_with("otel-metrics") || index_id.starts_with("metrics-")
-}
-
 #[macro_export]
 macro_rules! ignore_error_kind {
     ($kind:path, $expr:expr) => {
@@ -423,27 +418,6 @@ mod tests {
         assert_eq!(div_ceil_u32(0, 3), 0);
     }
 
-    #[test]
-    fn test_is_metrics_index() {
-        // OpenTelemetry metrics indexes
-        assert!(is_metrics_index("otel-metrics-v0_7"));
-        assert!(is_metrics_index("otel-metrics"));
-        assert!(is_metrics_index("otel-metrics-custom"));
-
-        // Generic metrics indexes
-        assert!(is_metrics_index("metrics-default"));
-        assert!(is_metrics_index("metrics-"));
-        assert!(is_metrics_index("metrics-my-app"));
-
-        // Non-metrics indexes
-        assert!(!is_metrics_index("otel-logs-v0_7"));
-        assert!(!is_metrics_index("otel-traces-v0_7"));
-        assert!(!is_metrics_index("my-index"));
-        assert!(!is_metrics_index("logs-default"));
-        assert!(!is_metrics_index("metrics")); // No hyphen after "metrics"
-        assert!(!is_metrics_index("my-metrics-index")); // Not prefixed
-    }
-
     #[test]
     fn test_parse_bool_lenient() {
         assert_eq!(parse_bool_lenient("true"), Some(true));

quickwit/quickwit-common/src/metrics_specific.rs

@@ -0,0 +1,31 @@
+/// Returns whether the given index ID corresponds to a metrics index.
+///
+/// Metrics indexes use the Parquet/DataFusion pipeline instead of the Tantivy pipeline.
+/// An index is considered a metrics index if it starts with "otel-metrics" or "metrics-".
+pub fn is_metrics_index(index_id: &str) -> bool {
+    index_id.starts_with("otel-metrics") || index_id.starts_with("metrics-")
+}
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn test_is_metrics_index() {
+        // OpenTelemetry metrics indexes
+        assert!(is_metrics_index("otel-metrics-v0_7"));
+        assert!(is_metrics_index("otel-metrics"));
+        assert!(is_metrics_index("otel-metrics-custom"));
+
+        // Generic metrics indexes
+        assert!(is_metrics_index("metrics-default"));
+        assert!(is_metrics_index("metrics-"));
+        assert!(is_metrics_index("metrics-my-app"));
+
+        // Non-metrics indexes
+        assert!(!is_metrics_index("otel-logs-v0_7"));
+        assert!(!is_metrics_index("otel-traces-v0_7"));
+        assert!(!is_metrics_index("my-index"));
+        assert!(!is_metrics_index("logs-default"));
+        assert!(!is_metrics_index("metrics")); // No hyphen after "metrics"
+        assert!(!is_metrics_index("my-metrics-index")); // Not prefixed
+    }
+}

quickwit/quickwit-indexing/Cargo.toml

@@ -50,7 +50,7 @@ utoipa = { workspace = true }
 vrl = { workspace = true, optional = true }
 warp = { workspace = true, optional = true }
 
-arrow = { workspace = true }
+arrow = { workspace = true, optional = true }
 
 quickwit-actors = { workspace = true }
 quickwit-aws = { workspace = true }
@@ -62,7 +62,7 @@ quickwit-doc-mapper = { workspace = true }
 quickwit-ingest = { workspace = true }
 quickwit-metastore = { workspace = true }
 quickwit-opentelemetry = { workspace = true }
-quickwit-parquet-engine = { workspace = true }
+quickwit-parquet-engine = { workspace = true, optional = true }
 quickwit-proto = { workspace = true }
 quickwit-storage = { workspace = true }
 
@@ -106,6 +106,7 @@ testsuite = [
   "quickwit-proto/testsuite",
   "quickwit-storage/testsuite"
 ]
+metrics = ["dep:arrow", "dep:quickwit-parquet-engine"]
 vrl = ["dep:vrl", "quickwit-config/vrl"]
 postgres = ["quickwit-metastore/postgres"]
 ci-test = []
@@ -114,7 +115,6 @@ ci-test = []
 bytes = { workspace = true }
 criterion = { workspace = true, features = ["async_tokio"] }
 mockall = { workspace = true }
-parquet = { workspace = true }
 proptest = { workspace = true }
 prost = { workspace = true }
 rand = { workspace = true }
@@ -123,8 +123,8 @@ sqlx = { workspace = true, features = ["runtime-tokio", "postgres"] }
 tempfile = { workspace = true }
 
 quickwit-actors = { workspace = true, features = ["testsuite"] }
-quickwit-parquet-engine = { workspace = true, features = ["testsuite"] }
 quickwit-cluster = { workspace = true, features = ["testsuite"] }
+quickwit-parquet-engine = { workspace = true, features = ["testsuite"] }
 quickwit-common = { workspace = true, features = ["testsuite"] }
 quickwit-config = { workspace = true, features = ["testsuite"] }
 quickwit-doc-mapper = { workspace = true, features = ["testsuite"] }

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

@@ -49,8 +49,8 @@ use tokio::sync::Semaphore;
 use tracing::{Span, info, info_span, warn};
 use ulid::Ulid;
 
-use crate::actors::IndexSerializer;
-use crate::actors::cooperative_indexing::{CooperativeIndexingCycle, CooperativeIndexingPeriod};
+use super::IndexSerializer;
+use super::cooperative_indexing::{CooperativeIndexingCycle, CooperativeIndexingPeriod};
 use crate::models::{
     CommitTrigger, EmptySplit, IndexedSplitBatchBuilder, IndexedSplitBuilder, NewPublishLock,
     NewPublishToken, ProcessedDoc, ProcessedDocBatch, PublishLock,
@@ -715,8 +715,7 @@ mod tests {
     use quickwit_proto::types::{IndexUid, NodeId, PipelineUid};
     use tantivy::{DateTime, doc};
 
-    use super::*;
-    use crate::actors::indexer::{IndexerCounters, record_timestamp};
+    use super::{IndexerCounters, record_timestamp, *};
 
     #[test]
     fn test_record_timestamp() {