read and coalescing metrics (#2413)

This PR introduces VortexMetrics, a thin wrapper on top of
witchcraft-metrics that supports default tags. I have added metrics for
`VortexReadAt`, and coalescing metrics to the scan driver.

Next step would be to tie these with the datafusion metrics on
`VortexExec::metrics`

Author: onursatici

Cargo.lock

@@ -18,6 +18,7 @@ members = [
     "vortex-ipc",
     "vortex-layout",
     "vortex-mask",
+    "vortex-metrics",
     "vortex-proto",
     "vortex-sampling-compressor",
     "vortex-scalar",
@@ -158,6 +159,7 @@ tracing-subscriber = "0.3.19"
 url = "2.5.4"
 uuid = "1.8.0"
 wasm-bindgen-futures = "0.4.39"
+witchcraft-metrics = "1.0.1"
 
 # BEGIN crates published by this project
 vortex = { version = "0.24.0", path = "./vortex" }
@@ -180,6 +182,7 @@ vortex-io = { version = "0.24.0", path = "./vortex-io" }
 vortex-ipc = { version = "0.24.0", path = "./vortex-ipc" }
 vortex-layout = { version = "0.24.0", path = "./vortex-layout" }
 vortex-mask = { version = "0.24.0", path = "./vortex-mask" }
+vortex-metrics = { version = "0.24.0", path = "./vortex-metrics" }
 vortex-proto = { version = "0.24.0", path = "./vortex-proto" }
 vortex-runend = { version = "0.24.0", path = "./encodings/runend" }
 vortex-sampling-compressor = { version = "0.24.0", path = "./vortex-sampling-compressor" }

Cargo.toml

@@ -48,6 +48,8 @@ vortex-expr = { workspace = true, features = ["datafusion"] }
 vortex-file = { workspace = true, features = ["object_store", "tokio"] }
 vortex-io = { workspace = true, features = ["object_store", "tokio"] }
 vortex-layout = { workspace = true, features = ["tokio"] }
+vortex-metrics = { workspace = true }
+
 [features]
 tracing = ["dep:tracing", "dep:tracing-futures", "vortex-io/tracing"]
 

vortex-datafusion/Cargo.toml

@@ -11,7 +11,8 @@ use vortex_error::VortexResult;
 use vortex_expr::{ExprRef, VortexExpr};
 use vortex_file::executor::{TaskExecutor, TokioExecutor};
 use vortex_file::{SplitBy, VortexOpenOptions};
-use vortex_io::ObjectStoreReadAt;
+use vortex_io::{InstrumentedReadAt, ObjectStoreReadAt};
+use vortex_metrics::VortexMetrics;
 
 use super::cache::FileLayoutCache;
 
@@ -54,10 +55,16 @@ impl VortexFileOpener {
 
 impl FileOpener for VortexFileOpener {
     fn open(&self, file_meta: FileMeta) -> DFResult<FileOpenFuture> {
-        let read_at = ObjectStoreReadAt::new(
-            self.object_store.clone(),
-            file_meta.location().clone(),
-            Some(self.scheme.clone()),
+        let metrics = VortexMetrics::default_with_tags(
+            [("filename", file_meta.location().to_string())].as_slice(),
+        );
+        let read_at = InstrumentedReadAt::new(
+            ObjectStoreReadAt::new(
+                self.object_store.clone(),
+                file_meta.location().clone(),
+                Some(self.scheme.clone()),
+            ),
+            &metrics,
         );
 
         let filter = self.filter.clone();

vortex-datafusion/src/persistent/opener.rs

@@ -36,6 +36,7 @@ vortex-expr = { workspace = true }
 vortex-flatbuffers = { workspace = true, features = ["file"] }
 vortex-io = { workspace = true }
 vortex-layout = { workspace = true }
+vortex-metrics = { workspace = true }
 vortex-sampling-compressor = { workspace = true }
 
 [dev-dependencies]

vortex-file/Cargo.toml

@@ -5,6 +5,7 @@ use vortex_array::stats::StatsSet;
 use vortex_array::ContextRef;
 use vortex_dtype::DType;
 use vortex_layout::scan::ScanBuilder;
+use vortex_metrics::VortexMetrics;
 
 use crate::footer::FileLayout;
 use crate::open::FileType;
@@ -16,6 +17,7 @@ pub struct VortexFile<F: FileType> {
     pub(crate) ctx: ContextRef,
     pub(crate) file_layout: FileLayout,
     pub(crate) segment_cache: Arc<dyn SegmentCache>,
+    pub(crate) metrics: Arc<VortexMetrics>,
     pub(crate) _marker: PhantomData<F>,
 }
 
@@ -42,6 +44,7 @@ impl<F: FileType> VortexFile<F> {
             self.options.clone(),
             self.file_layout.clone(),
             self.segment_cache.clone(),
+            self.metrics.clone(),
         );
         ScanBuilder::new(
             driver,

vortex-file/src/file.rs

@@ -10,6 +10,7 @@ use vortex_error::{vortex_err, vortex_panic, VortexExpect, VortexResult};
 use vortex_io::VortexReadAt;
 use vortex_layout::scan::ScanDriver;
 use vortex_layout::segments::{AsyncSegmentReader, SegmentId};
+use vortex_metrics::{Counter, MetricId, VortexMetrics};
 
 use crate::footer::{FileLayout, Segment};
 use crate::segments::channel::SegmentChannel;
@@ -31,13 +32,15 @@ impl<R: VortexReadAt> FileType for GenericVortexFile<R> {
         options: Self::Options,
         file_layout: FileLayout,
         segment_cache: Arc<dyn SegmentCache>,
+        metrics: Arc<VortexMetrics>,
     ) -> Self::ScanDriver {
         GenericScanDriver {
             read,
             options,
             file_layout,
             segment_cache,
             segment_channel: SegmentChannel::new(),
+            metrics: metrics.into(),
         }
     }
 }
@@ -68,6 +71,7 @@ pub struct GenericScanDriver<R> {
     file_layout: FileLayout,
     segment_cache: Arc<dyn SegmentCache>,
     segment_channel: SegmentChannel,
+    metrics: CoalescingMetrics,
 }
 
 impl<R: VortexReadAt> ScanDriver for GenericScanDriver<R> {
@@ -145,7 +149,8 @@ impl<R: VortexReadAt> ScanDriver for GenericScanDriver<R> {
         let perf_hint = self.read.performance_hint();
         let io_stream = io_stream
             .map(move |r| coalesce(r, perf_hint.coalescing_window(), perf_hint.max_read()))
-            .flat_map(stream::iter);
+            .flat_map(stream::iter)
+            .inspect(move |coalesced| self.metrics.record(coalesced));
 
         // Submit the coalesced requests to the I/O.
         let read = self.read.clone();
@@ -188,6 +193,10 @@ impl FileSegmentRequest {
             .map_err(|_| vortex_err!("send failed"))
             .vortex_expect("send failed");
     }
+
+    fn range(&self) -> Range<u64> {
+        self.location.offset..self.location.offset + self.location.length as u64
+    }
 }
 
 #[derive(Debug)]
@@ -271,9 +280,7 @@ fn coalesce(
     max_read: Option<u64>,
 ) -> Vec<CoalescedSegmentRequest> {
     let fetch_ranges = merge_ranges(
-        requests
-            .iter()
-            .map(|r| r.location.offset..r.location.offset + r.location.length as u64),
+        requests.iter().map(|r| r.range()),
         coalescing_window,
         max_read,
     );
@@ -348,6 +355,48 @@ where
     ret
 }
 
+struct CoalescingMetrics {
+    bytes_uncoalesced: Arc<Counter>,
+    bytes_coalesced: Arc<Counter>,
+    request_count_uncoalesced: Arc<Counter>,
+    request_count_coalesced: Arc<Counter>,
+}
+
+impl From<Arc<VortexMetrics>> for CoalescingMetrics {
+    fn from(metrics: Arc<VortexMetrics>) -> Self {
+        let byte_ranges = MetricId::new("vortex.scan.requests.bytes");
+        let requests = MetricId::new("vortex.scan.requests.count");
+        Self {
+            bytes_uncoalesced: metrics.counter(byte_ranges.clone().with_tag("kind", "uncoalesced")),
+            bytes_coalesced: metrics.counter(byte_ranges.with_tag("kind", "coalesced")),
+            request_count_uncoalesced: metrics
+                .counter(requests.clone().with_tag("kind", "uncoalesced")),
+            request_count_coalesced: metrics.counter(requests.with_tag("kind", "coalesced")),
+        }
+    }
+}
+
+impl CoalescingMetrics {
+    fn record(&self, req: &CoalescedSegmentRequest) {
+        // record request counts
+        self.request_count_coalesced.inc();
+        if let Ok(len) = req.requests.len().try_into() {
+            self.request_count_uncoalesced.add(len);
+        }
+
+        // record uncoalesced total byte requests vs coalesced
+        if let Ok(bytes) = (req.byte_range.end - req.byte_range.start).try_into() {
+            self.bytes_coalesced.add(bytes);
+        }
+        self.bytes_uncoalesced.add(
+            req.requests
+                .iter()
+                .map(|req| req.location.length as i64)
+                .sum(),
+        );
+    }
+}
+
 #[cfg(test)]
 mod test {
     use super::*;

vortex-file/src/generic.rs

@@ -6,6 +6,7 @@ use vortex_buffer::ByteBuffer;
 use vortex_error::{vortex_err, VortexResult};
 use vortex_layout::scan::ScanDriver;
 use vortex_layout::segments::{AsyncSegmentReader, SegmentId};
+use vortex_metrics::VortexMetrics;
 
 use crate::segments::SegmentCache;
 use crate::{FileLayout, FileType, Segment};
@@ -30,6 +31,7 @@ impl FileType for InMemoryVortexFile {
         _options: Self::Options,
         file_layout: FileLayout,
         _segment_cache: Arc<dyn SegmentCache>,
+        _metrics: Arc<VortexMetrics>,
     ) -> Self::ScanDriver {
         Self {
             buffer: read,