From 49463d67eb75b627c7da03fd733b4b0ce588b6fa Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 22 Aug 2026 18:03:57 +0000 Subject: [PATCH 1/3] Add measured decompression throughput to the array tree display The tree display already answers "where do the bytes go". This adds a composable extractor that answers "where does the decompression time go", rendered through the same tree. `DecompressionProfile::measure` canonicalizes every node of an encoding tree in isolation and records the median wall time, keyed by node identity, so `ThroughputExtractor` is a pure lookup on the tree `TreeDisplay` already walks. No change to the renderer or to the `TreeDisplayExtractor` contract. A node's self time is its subtree time minus its children's. Encodings that rewrite a `(parent, child)` pair through an `execute_parent` kernel or a `reduce_parent` rule reach canonical form without canonicalizing the child, so they cost less than the sum of their children. That is reported as a fusion saving rather than a negative self time, which needs no executor instrumentation: it falls out of comparing a node against its children. Gated behind the off-by-default `profile-throughput` feature, since the measurement uses `Instant` and performs O(nodes * reps) decompressions. Running the `decompress` benchmark's trees with `VORTEX_DECOMPRESS_PROFILE=1` prints the per-subtree breakdown behind each of that benchmark's totals. Signed-off-by: Claude Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_016uJfeYsetVkEKGVFGuMS6a --- vortex-array/Cargo.toml | 2 + vortex-array/src/display/extractors/mod.rs | 4 + .../src/display/extractors/throughput.rs | 129 +++++++ vortex-array/src/display/mod.rs | 40 +++ vortex-array/src/display/profile.rs | 331 ++++++++++++++++++ vortex/Cargo.toml | 4 +- .../common_encoding_tree_throughput.rs | 34 +- 7 files changed, 539 insertions(+), 5 deletions(-) create mode 100644 vortex-array/src/display/extractors/throughput.rs create mode 100644 vortex-array/src/display/profile.rs diff --git a/vortex-array/Cargo.toml b/vortex-array/Cargo.toml index 7c5936627e5..e330e5421c8 100644 --- a/vortex-array/Cargo.toml +++ b/vortex-array/Cargo.toml @@ -76,6 +76,8 @@ vortex-utils = { workspace = true, features = ["dyn-traits"] } arbitrary = ["dep:arbitrary", "dep:primitive-types"] canonical_counter = [] cudarc = ["dep:cudarc"] +# Measured per-subtree decompression throughput for the array tree display. +profile-throughput = [] table-display = ["dep:tabled"] _test-harness = ["dep:goldenfile", "dep:rstest", "dep:rstest_reuse"] serde = ["dep:serde", "vortex-buffer/serde", "vortex-mask/serde"] diff --git a/vortex-array/src/display/extractors/mod.rs b/vortex-array/src/display/extractors/mod.rs index 754ac16a57a..088fee924bc 100644 --- a/vortex-array/src/display/extractors/mod.rs +++ b/vortex-array/src/display/extractors/mod.rs @@ -6,9 +6,13 @@ mod encoding_summary; mod metadata; mod nbytes; mod stats; +#[cfg(feature = "profile-throughput")] +mod throughput; pub use buffer::BufferExtractor; pub use encoding_summary::EncodingSummaryExtractor; pub use metadata::MetadataExtractor; pub use nbytes::NbytesExtractor; pub use stats::StatsExtractor; +#[cfg(feature = "profile-throughput")] +pub use throughput::ThroughputExtractor; diff --git a/vortex-array/src/display/extractors/throughput.rs b/vortex-array/src/display/extractors/throughput.rs new file mode 100644 index 00000000000..163b8988185 --- /dev/null +++ b/vortex-array/src/display/extractors/throughput.rs @@ -0,0 +1,129 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +use std::fmt; +use std::time::Duration; + +use vortex_error::VortexResult; +use vortex_session::VortexSession; + +use crate::ArrayRef; +use crate::display::extractor::TreeContext; +use crate::display::extractor::TreeExtractor; +use crate::display::profile::DecompressionProfile; +use crate::display::profile::NodeTiming; +use crate::display::profile::ProfileOptions; + +/// Extractor that adds a `throughput:` detail line from a measured [`DecompressionProfile`]. +/// +/// The line reports the time to canonicalize the subtree, its share of the whole tree's time, the +/// rates that time implies, and either the node's self time or the amount of child work it fuses +/// into itself. +pub struct ThroughputExtractor { + profile: DecompressionProfile, +} + +impl ThroughputExtractor { + /// Annotate a tree with an already-measured profile. + pub fn new(profile: DecompressionProfile) -> Self { + Self { profile } + } + + /// Measure `array` and annotate it with the result. + pub fn measure( + array: &ArrayRef, + session: &VortexSession, + options: ProfileOptions, + ) -> VortexResult { + Ok(Self::new(DecompressionProfile::measure( + array, session, options, + )?)) + } + + /// The profile backing this extractor. + pub fn profile(&self) -> &DecompressionProfile { + &self.profile + } +} + +impl TreeExtractor for ThroughputExtractor { + fn write_details( + &self, + array: &ArrayRef, + _ctx: &TreeContext, + f: &mut crate::display::IndentedFormatter<'_, '_>, + ) -> fmt::Result { + let Some(timing) = self.profile.get(array) else { + return Ok(()); + }; + let (indent, f) = f.parts(); + write!( + f, + "{indent}throughput: {}", + Timing(timing, self.profile.root_time()) + )?; + writeln!(f) + } +} + +struct Timing<'a>(&'a NodeTiming, Duration); + +impl fmt::Display for Timing<'_> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let Self(timing, root) = *self; + let percent = if root.is_zero() { + 0.0 + } else { + 100_f64 * timing.subtree.as_secs_f64() / root.as_secs_f64() + }; + write!( + f, + "{} ({percent:.2}%) | in {} | out {} | {}", + Elapsed(timing.subtree), + Rate(timing.input_bytes_per_sec(), &["B", "kB", "MB", "GB"]), + Rate(timing.output_bytes_per_sec(), &["B", "kB", "MB", "GB"]), + Rate(timing.rows_per_sec(), &["row", "krow", "Mrow", "Grow"]), + )?; + match timing.fusion_saving() { + Some(saving) => write!(f, " | fuses children (saves {})", Elapsed(saving)), + None => write!(f, " | self {}", Elapsed(timing.self_time())), + } + } +} + +/// A duration, rendered as `1.81ms`. +struct Elapsed(Duration); + +impl fmt::Display for Elapsed { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let secs = self.0.as_secs_f64(); + for (scale, unit) in [(1.0, "s"), (1e-3, "ms"), (1e-6, "\u{b5}s")] { + if secs >= scale { + return write!(f, "{:.2}{unit}", secs / scale); + } + } + write!(f, "{:.0}ns", secs * 1e9) + } +} + +/// A per-second rate, rendered in the largest unit that keeps it above one, e.g. `1.90 GB/s`. +struct Rate(f64, &'static [&'static str]); + +impl fmt::Display for Rate { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let Self(rate, units) = *self; + if !rate.is_finite() { + return write!(f, "n/a"); + } + let mut scale = 1.0; + let mut unit = units[0]; + for next in &units[1..] { + if rate < scale * 1e3 { + break; + } + scale *= 1e3; + unit = next; + } + write!(f, "{:.2} {unit}/s", rate / scale) + } +} diff --git a/vortex-array/src/display/mod.rs b/vortex-array/src/display/mod.rs index 780140af2ac..4fd6d485a8f 100644 --- a/vortex-array/src/display/mod.rs +++ b/vortex-array/src/display/mod.rs @@ -3,6 +3,8 @@ mod extractor; mod extractors; +#[cfg(feature = "profile-throughput")] +pub mod profile; mod tree_display; use std::fmt::Display; @@ -16,6 +18,8 @@ pub use extractors::EncodingSummaryExtractor; pub use extractors::MetadataExtractor; pub use extractors::NbytesExtractor; pub use extractors::StatsExtractor; +#[cfg(feature = "profile-throughput")] +pub use extractors::ThroughputExtractor; use itertools::Itertools as _; pub use tree_display::TreeDisplay; @@ -430,6 +434,42 @@ impl ArrayRef { TreeDisplay::default_display(self.clone()) } + /// Display the tree of encodings annotated with measured decompression throughput. + /// + /// Every node is canonicalized in isolation `warmup + reps` times (see [`ProfileOptions`]), + /// so this is a profiling call rather than a formatting one. Each node reports the time to + /// decode its own subtree, that time's share of the whole tree, the rates it implies, and + /// either its self time or how much child work it fuses into itself. + /// + /// [`ProfileOptions`]: profile::ProfileOptions + /// + /// # Examples + /// ``` + /// # use vortex_array::IntoArray; + /// # use vortex_array::array_session; + /// # use vortex_array::display::profile::ProfileOptions; + /// # use vortex_buffer::buffer; + /// let array = buffer![0_i16, 1, 2, 3, 4].into_array(); + /// let tree = array + /// .display_tree_throughput(&array_session(), ProfileOptions::default())? + /// .to_string(); + /// assert!(tree.starts_with("root: vortex.primitive(i16, len=5)")); + /// assert!(tree.contains("throughput: ")); + /// # Ok::<(), vortex_error::VortexError>(()) + /// ``` + #[cfg(feature = "profile-throughput")] + pub fn display_tree_throughput( + &self, + session: &vortex_session::VortexSession, + options: profile::ProfileOptions, + ) -> vortex_error::VortexResult { + Ok(self + .tree_display_builder() + .with(EncodingSummaryExtractor) + .with(NbytesExtractor) + .with(ThroughputExtractor::measure(self, session, options)?)) + } + /// Create a tree display with all built-in extractors (nbytes, stats, metadata, buffers). /// /// This is the default, fully-detailed tree display. Use diff --git a/vortex-array/src/display/profile.rs b/vortex-array/src/display/profile.rs new file mode 100644 index 00000000000..7ed59d969e9 --- /dev/null +++ b/vortex-array/src/display/profile.rs @@ -0,0 +1,331 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Measured decompression cost for every subtree of an array. +//! +//! [`DecompressionProfile::measure`] canonicalizes each node of an encoding tree in isolation and +//! records how long it took. The result is keyed by node identity so that +//! [`ThroughputExtractor`](crate::display::ThroughputExtractor) can annotate the same tree that +//! [`TreeDisplay`](crate::display::TreeDisplay) already renders. +//! +//! # Fusion +//! +//! A node's *subtree* time is the wall time to drive that node to canonical form on its own. Its +//! *self* time is the subtree time minus the subtree times of its children, i.e. the work the node +//! performs beyond decoding what it is built from. +//! +//! Encodings do not always pay for their children. The executor rewrites `(parent, child)` pairs +//! through `execute_parent` kernels and `reduce_parent` rules, so a parent may reach canonical form +//! without ever canonicalizing its child. Such a node costs *less* than the sum of its children, +//! which this module reports as a [`NodeTiming::fusion_saving`] rather than a negative self time. +//! Measuring fusion this way needs no executor instrumentation: it falls out of comparing a node's +//! own cost against its children's. + +use std::time::Duration; +use std::time::Instant; + +use vortex_error::VortexResult; +use vortex_session::VortexSession; +use vortex_utils::aliases::hash_map::HashMap; + +use crate::ArrayRef; +use crate::Canonical; +use crate::IntoArray as _; +use crate::VortexSessionExecute as _; + +/// How many times to canonicalize each node when building a [`DecompressionProfile`]. +#[derive(Debug, Clone, Copy)] +pub struct ProfileOptions { + /// Untimed canonicalizations run before measurement, to warm caches and lazily-computed stats. + pub warmup: usize, + /// Timed canonicalizations. The reported time is the median. + pub reps: usize, +} + +impl Default for ProfileOptions { + fn default() -> Self { + Self { warmup: 1, reps: 5 } + } +} + +/// The measured decompression cost of a single node. +#[derive(Debug, Clone, Copy)] +pub struct NodeTiming { + /// Median time to canonicalize this node, including everything below it. + pub subtree: Duration, + /// Sum of the [`Self::subtree`] times of this node's direct children. + pub children: Duration, + /// Compressed size of this subtree, as reported by [`ArrayRef::nbytes`]. + pub input_nbytes: u64, + /// Size of the canonical array this subtree decodes into. + pub output_nbytes: u64, + /// Logical length of this node. + pub rows: u64, +} + +impl NodeTiming { + /// Time this node spends beyond decoding its children. + /// + /// Zero when the node costs less than its children, which is reported by + /// [`Self::fusion_saving`] instead. + pub fn self_time(&self) -> Duration { + self.subtree.saturating_sub(self.children) + } + + /// How much cheaper this node is than decoding its children separately. + /// + /// `Some` only when the node fuses its children's decompression into its own work. + pub fn fusion_saving(&self) -> Option { + let saving = self.children.saturating_sub(self.subtree); + (!saving.is_zero()).then_some(saving) + } + + /// Compressed bytes consumed per second while canonicalizing this subtree. + pub fn input_bytes_per_sec(&self) -> f64 { + per_sec(self.input_nbytes, self.subtree) + } + + /// Canonical bytes produced per second while canonicalizing this subtree. + pub fn output_bytes_per_sec(&self) -> f64 { + per_sec(self.output_nbytes, self.subtree) + } + + /// Rows produced per second while canonicalizing this subtree. + pub fn rows_per_sec(&self) -> f64 { + per_sec(self.rows, self.subtree) + } +} + +fn per_sec(amount: u64, elapsed: Duration) -> f64 { + let secs = elapsed.as_secs_f64(); + if secs <= 0.0 { + return f64::INFINITY; + } + amount as f64 / secs +} + +/// Decompression timings for every node of an encoding tree. +/// +/// Build one with [`Self::measure`], then render it with +/// [`ArrayRef::display_tree_throughput`]. +#[derive(Debug, Clone)] +pub struct DecompressionProfile { + root: Duration, + nodes: HashMap, +} + +impl DecompressionProfile { + /// Measure the decompression cost of `array` and of every node beneath it. + /// + /// Each node is canonicalized `warmup + reps` times, so this performs `O(nodes * reps)` + /// decompressions. It is a profiling entry point, never something a `Display` implementation + /// should reach for. + pub fn measure( + array: &ArrayRef, + session: &VortexSession, + options: ProfileOptions, + ) -> VortexResult { + let mut profile = Self { + root: Duration::ZERO, + nodes: HashMap::default(), + }; + profile.root = profile.measure_node(array, session, options)?; + Ok(profile) + } + + /// The timing recorded for `array`, if it was part of the measured tree. + pub fn get(&self, array: &ArrayRef) -> Option<&NodeTiming> { + self.nodes.get(&array.addr()) + } + + /// The time taken to canonicalize the whole tree, used as the denominator for percentages. + pub fn root_time(&self) -> Duration { + self.root + } + + /// The number of measured nodes. Repeated occurrences of one shared node count once. + pub fn len(&self) -> usize { + self.nodes.len() + } + + /// Whether no node was measured. + pub fn is_empty(&self) -> bool { + self.nodes.is_empty() + } + + /// Measure `array` and its children, returning `array`'s subtree time. + fn measure_node( + &mut self, + array: &ArrayRef, + session: &VortexSession, + options: ProfileOptions, + ) -> VortexResult { + let mut children = Duration::ZERO; + for child in array.children() { + children += self.measure_node(&child, session, options)?; + } + + let (subtree, output_nbytes) = time_canonicalize(array, session, options)?; + self.nodes.insert( + array.addr(), + NodeTiming { + subtree, + children, + input_nbytes: array.nbytes(), + output_nbytes, + rows: array.len() as u64, + }, + ); + Ok(subtree) + } +} + +/// Canonicalize `array` repeatedly, returning the median elapsed time and the canonical size. +fn time_canonicalize( + array: &ArrayRef, + session: &VortexSession, + options: ProfileOptions, +) -> VortexResult<(Duration, u64)> { + let mut output_nbytes = 0; + for _ in 0..options.warmup { + let mut ctx = session.create_execution_ctx(); + output_nbytes = array + .clone() + .execute::(&mut ctx)? + .into_array() + .nbytes(); + } + + let mut elapsed = Vec::with_capacity(options.reps); + for _ in 0..options.reps { + let mut ctx = session.create_execution_ctx(); + let array = array.clone(); + let start = Instant::now(); + let canonical = array.execute::(&mut ctx)?; + elapsed.push(start.elapsed()); + output_nbytes = canonical.into_array().nbytes(); + } + + elapsed.sort_unstable(); + Ok(( + elapsed.get(elapsed.len() / 2).copied().unwrap_or_default(), + output_nbytes, + )) +} + +#[cfg(test)] +mod tests { + // The profile is session-agnostic; these tests do not need a configured session. + #![allow(clippy::disallowed_methods)] + + use vortex_buffer::buffer; + use vortex_error::VortexExpect as _; + + use super::*; + use crate::arrays::DictArray; + use crate::display::ThroughputExtractor; + use crate::legacy_session; + + /// Options that keep the tests fast: the timings themselves are not asserted on. + fn fast() -> ProfileOptions { + ProfileOptions { warmup: 0, reps: 1 } + } + + fn dict_array() -> VortexResult { + Ok(DictArray::try_new( + buffer![0u32, 1, 0, 1, 2].into_array(), + buffer![10i32, 20, 30].into_array(), + )? + .into_array()) + } + + #[test] + fn measures_every_node() -> VortexResult<()> { + let array = dict_array()?; + let profile = DecompressionProfile::measure(&array, legacy_session(), fast())?; + + assert_eq!(profile.len(), 3, "root plus codes plus values"); + for node in [array.clone()].into_iter().chain(array.children()) { + assert!(profile.get(&node).is_some(), "missing timing for {node}"); + } + Ok(()) + } + + #[test] + fn child_timings_roll_up_into_the_parent() -> VortexResult<()> { + let array = dict_array()?; + let profile = DecompressionProfile::measure(&array, legacy_session(), fast())?; + + let root = profile.get(&array).vortex_expect("root is measured"); + let children: Duration = array + .children() + .iter() + .map(|child| { + profile + .get(child) + .vortex_expect("child is measured") + .subtree + }) + .sum(); + assert_eq!(root.children, children); + assert_eq!(root.rows, array.len() as u64); + assert_eq!(root.input_nbytes, array.nbytes()); + assert_eq!(profile.root_time(), root.subtree); + + // Self time and fusion saving are two directions of the same comparison, never both. + assert!(root.self_time().is_zero() || root.fusion_saving().is_none()); + Ok(()) + } + + #[test] + fn leaf_has_no_children_and_no_fusion() -> VortexResult<()> { + let array = buffer![0i32, 1, 2].into_array(); + let profile = DecompressionProfile::measure(&array, legacy_session(), fast())?; + + let leaf = profile.get(&array).vortex_expect("root is measured"); + assert_eq!(leaf.children, Duration::ZERO); + assert_eq!(leaf.self_time(), leaf.subtree); + assert_eq!(leaf.fusion_saving(), None); + Ok(()) + } + + #[test] + fn every_rendered_node_carries_a_throughput_line() -> VortexResult<()> { + let array = dict_array()?; + let rendered = array + .tree_display_builder() + .with(crate::display::EncodingSummaryExtractor) + .with(ThroughputExtractor::measure( + &array, + legacy_session(), + fast(), + )?) + .to_string(); + + assert_eq!( + rendered.matches("throughput: ").count(), + 3, + "one line per node in:\n{rendered}" + ); + assert!(rendered.contains(" | in "), "{rendered}"); + assert!(rendered.contains(" | out "), "{rendered}"); + Ok(()) + } + + #[test] + fn unmeasured_nodes_are_skipped() -> VortexResult<()> { + // A profile of one array must not annotate an unrelated tree. + let measured = buffer![0i32, 1, 2].into_array(); + let other = buffer![9i64, 8].into_array(); + let profile = DecompressionProfile::measure(&measured, legacy_session(), fast())?; + + assert!(profile.get(&other).is_none()); + let rendered = other + .tree_display_builder() + .with(crate::display::EncodingSummaryExtractor) + .with(ThroughputExtractor::new(profile)) + .to_string(); + assert!(!rendered.contains("throughput"), "{rendered}"); + Ok(()) + } +} diff --git a/vortex/Cargo.toml b/vortex/Cargo.toml index fc3f9e420f8..178f8e02e55 100644 --- a/vortex/Cargo.toml +++ b/vortex/Cargo.toml @@ -67,7 +67,7 @@ serde_json = { workspace = true } tokio = { workspace = true, features = ["full"] } tracing = { workspace = true } tracing-subscriber = { workspace = true } -vortex = { path = ".", features = ["tokio"] } +vortex = { path = ".", features = ["profile-throughput", "tokio"] } [features] default = ["files", "wasm-bindgen", "zstd"] @@ -94,6 +94,8 @@ wasm-bindgen = [ "vortex-layout/wasm-bindgen", ] pretty = ["vortex-array/table-display"] +# Measured per-subtree decompression throughput for the array tree display. +profile-throughput = ["vortex-array/profile-throughput"] serde = ["vortex-array/serde", "vortex-buffer/serde", "vortex-mask/serde"] # Exposes experimental row-function APIs without compatibility guarantees. unstable_row_fns = ["vortex-array/unstable_row_fns"] diff --git a/vortex/benches/common_encoding_tree_throughput.rs b/vortex/benches/common_encoding_tree_throughput.rs index 2032eb3467a..2697c496127 100644 --- a/vortex/benches/common_encoding_tree_throughput.rs +++ b/vortex/benches/common_encoding_tree_throughput.rs @@ -26,6 +26,7 @@ use vortex::array::arrays::VarBinArray; use vortex::array::arrays::VarBinViewArray; use vortex::array::arrays::varbin::VarBinArraySlotsExt; use vortex::array::builtins::ArrayBuiltins; +use vortex::array::display::profile::ProfileOptions; use vortex::dtype::DType; use vortex::dtype::PType; use vortex::encodings::alp::ALP; @@ -54,6 +55,10 @@ static GLOBAL: MiMalloc = MiMalloc; fn main() { LazyLock::force(&SESSION); + if std::env::var_os("VORTEX_DECOMPRESS_PROFILE").is_some() { + print_decompress_profiles(); + return; + } divan::main(); } @@ -442,9 +447,9 @@ macro_rules! setup_fn { }; } -/// Benchmark decompression of various encoding trees -#[divan::bench( - args = [ +/// The encoding trees exercised by the `decompress` benchmark. +static DECOMPRESS_TREES: LazyLock<[SetupFn; 7]> = LazyLock::new(|| { + [ setup_fn!(setup::for_bp_u64), setup_fn!(setup::alp_for_bp_f64), setup_fn!(setup::dict_varbinview_string), @@ -453,7 +458,28 @@ macro_rules! setup_fn { setup_fn!(setup::dict_fsst_varbin_bp_string), setup_fn!(setup::datetime_for_bp), ] -)] +}); + +/// Print where each tree's decompression time goes, per subtree. +/// +/// This shares its trees with the `decompress` benchmark below, so the per-node timings can be +/// read against that benchmark's totals. Run with `VORTEX_DECOMPRESS_PROFILE=1`. +fn print_decompress_profiles() { + let options = ProfileOptions { + warmup: 3, + reps: 15, + }; + for setup_fn in DECOMPRESS_TREES.iter() { + let compressed = (setup_fn.func)(); + let tree = compressed + .display_tree_throughput(&SESSION, options) + .unwrap(); + println!("== {setup_fn} ==\n{tree}"); + } +} + +/// Benchmark decompression of various encoding trees +#[divan::bench(args = DECOMPRESS_TREES.iter().copied())] fn decompress(bencher: Bencher, setup_fn: SetupFn) { let compressed = setup_fn(); let nbytes = compressed.nbytes(); From 0cf8697bf3aaafd40e6c645b4d98fcc9ec210daa Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 24 Aug 2026 09:57:00 +0000 Subject: [PATCH 2/3] Polish the throughput extractor for review - Cover the duration and rate formatting with unit tests, and the zero-reps edge with a profile test. - Treat `reps: 0` as one timed run rather than reporting a zero time. - Document that a share above 100% is the fusion signal, that unprofiled nodes are skipped, and how shared subtrees are counted. - Run the crate's tests with `profile-throughput` on by default. Signed-off-by: Claude Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_016uJfeYsetVkEKGVFGuMS6a --- vortex-array/Cargo.toml | 1 + .../src/display/extractors/throughput.rs | 45 ++++++++++++++++--- vortex-array/src/display/profile.rs | 27 ++++++++--- 3 files changed, 60 insertions(+), 13 deletions(-) diff --git a/vortex-array/Cargo.toml b/vortex-array/Cargo.toml index e330e5421c8..89efc6427d3 100644 --- a/vortex-array/Cargo.toml +++ b/vortex-array/Cargo.toml @@ -96,6 +96,7 @@ serde_test = { workspace = true } test-with = { workspace = true } vortex-array = { path = ".", features = [ "_test-harness", + "profile-throughput", "table-display", "unstable_row_fns", ] } diff --git a/vortex-array/src/display/extractors/throughput.rs b/vortex-array/src/display/extractors/throughput.rs index 163b8988185..cfd80b4e116 100644 --- a/vortex-array/src/display/extractors/throughput.rs +++ b/vortex-array/src/display/extractors/throughput.rs @@ -8,6 +8,7 @@ use vortex_error::VortexResult; use vortex_session::VortexSession; use crate::ArrayRef; +use crate::display::IndentedFormatter; use crate::display::extractor::TreeContext; use crate::display::extractor::TreeExtractor; use crate::display::profile::DecompressionProfile; @@ -18,7 +19,11 @@ use crate::display::profile::ProfileOptions; /// /// The line reports the time to canonicalize the subtree, its share of the whole tree's time, the /// rates that time implies, and either the node's self time or the amount of child work it fuses -/// into itself. +/// into itself. A share above 100% means the child costs more on its own than the parent that +/// fuses it. +/// +/// Nodes missing from the profile are left unannotated, so a profile may be rendered against a +/// subtree of the tree it was measured on. pub struct ThroughputExtractor { profile: DecompressionProfile, } @@ -51,18 +56,17 @@ impl TreeExtractor for ThroughputExtractor { &self, array: &ArrayRef, _ctx: &TreeContext, - f: &mut crate::display::IndentedFormatter<'_, '_>, + f: &mut IndentedFormatter<'_, '_>, ) -> fmt::Result { let Some(timing) = self.profile.get(array) else { return Ok(()); }; let (indent, f) = f.parts(); - write!( + writeln!( f, "{indent}throughput: {}", Timing(timing, self.profile.root_time()) - )?; - writeln!(f) + ) } } @@ -97,7 +101,7 @@ struct Elapsed(Duration); impl fmt::Display for Elapsed { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { let secs = self.0.as_secs_f64(); - for (scale, unit) in [(1.0, "s"), (1e-3, "ms"), (1e-6, "\u{b5}s")] { + for (scale, unit) in [(1.0, "s"), (1e-3, "ms"), (1e-6, "µs")] { if secs >= scale { return write!(f, "{:.2}{unit}", secs / scale); } @@ -127,3 +131,32 @@ impl fmt::Display for Rate { write!(f, "{:.2} {unit}/s", rate / scale) } } + +#[cfg(test)] +mod tests { + use rstest::rstest; + + use super::*; + + #[rstest] + #[case(Duration::from_nanos(0), "0ns")] + #[case(Duration::from_nanos(640), "640ns")] + #[case(Duration::from_nanos(1_500), "1.50µs")] + #[case(Duration::from_micros(1_810), "1.81ms")] + #[case(Duration::from_millis(2_500), "2.50s")] + fn elapsed_picks_a_unit(#[case] elapsed: Duration, #[case] expected: &str) { + assert_eq!(Elapsed(elapsed).to_string(), expected); + } + + #[rstest] + #[case(0.0, "0.00 B/s")] + #[case(999.0, "999.00 B/s")] + #[case(1_000.0, "1.00 kB/s")] + #[case(1.9e9, "1.90 GB/s")] + // Rates beyond the largest unit keep that unit rather than wrapping around. + #[case(2e12, "2000.00 GB/s")] + #[case(f64::INFINITY, "n/a")] + fn rate_picks_a_unit(#[case] rate: f64, #[case] expected: &str) { + assert_eq!(Rate(rate, &["B", "kB", "MB", "GB"]).to_string(), expected); + } +} diff --git a/vortex-array/src/display/profile.rs b/vortex-array/src/display/profile.rs index 7ed59d969e9..f15f4857319 100644 --- a/vortex-array/src/display/profile.rs +++ b/vortex-array/src/display/profile.rs @@ -38,7 +38,8 @@ use crate::VortexSessionExecute as _; pub struct ProfileOptions { /// Untimed canonicalizations run before measurement, to warm caches and lazily-computed stats. pub warmup: usize, - /// Timed canonicalizations. The reported time is the median. + /// Timed canonicalizations. The reported time is the median. Values below one are treated as + /// one, since a node with no timed run has no time to report. pub reps: usize, } @@ -120,6 +121,10 @@ impl DecompressionProfile { /// Each node is canonicalized `warmup + reps` times, so this performs `O(nodes * reps)` /// decompressions. It is a profiling entry point, never something a `Display` implementation /// should reach for. + /// + /// Nodes are identified by the array they hold, so a subtree reachable by more than one path + /// is measured once per occurrence but recorded once. Its cost still counts towards each + /// parent that reaches it. pub fn measure( array: &ArrayRef, session: &VortexSession, @@ -196,8 +201,9 @@ fn time_canonicalize( .nbytes(); } - let mut elapsed = Vec::with_capacity(options.reps); - for _ in 0..options.reps { + let reps = options.reps.max(1); + let mut elapsed = Vec::with_capacity(reps); + for _ in 0..reps { let mut ctx = session.create_execution_ctx(); let array = array.clone(); let start = Instant::now(); @@ -207,10 +213,7 @@ fn time_canonicalize( } elapsed.sort_unstable(); - Ok(( - elapsed.get(elapsed.len() / 2).copied().unwrap_or_default(), - output_nbytes, - )) + Ok((elapsed[elapsed.len() / 2], output_nbytes)) } #[cfg(test)] @@ -277,6 +280,16 @@ mod tests { Ok(()) } + #[test] + fn zero_reps_still_reports_a_time() -> VortexResult<()> { + let array = buffer![0i32, 1, 2].into_array(); + let options = ProfileOptions { warmup: 0, reps: 0 }; + let profile = DecompressionProfile::measure(&array, legacy_session(), options)?; + + assert!(profile.get(&array).is_some()); + Ok(()) + } + #[test] fn leaf_has_no_children_and_no_fusion() -> VortexResult<()> { let array = buffer![0i32, 1, 2].into_array(); From 023be0d6286025d0d6d89a19b8913c9dadb2bdea Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 24 Aug 2026 10:06:00 +0000 Subject: [PATCH 3/3] Reword ProfileOptions docs to satisfy the typo check `typos` rejects "canonicalizations"; say "runs" instead, which reads better in both fields anyway. Signed-off-by: Claude Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_016uJfeYsetVkEKGVFGuMS6a --- vortex-array/src/display/profile.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/vortex-array/src/display/profile.rs b/vortex-array/src/display/profile.rs index f15f4857319..0adcf218d6d 100644 --- a/vortex-array/src/display/profile.rs +++ b/vortex-array/src/display/profile.rs @@ -36,10 +36,10 @@ use crate::VortexSessionExecute as _; /// How many times to canonicalize each node when building a [`DecompressionProfile`]. #[derive(Debug, Clone, Copy)] pub struct ProfileOptions { - /// Untimed canonicalizations run before measurement, to warm caches and lazily-computed stats. + /// Untimed runs performed before measurement, to warm caches and lazily-computed stats. pub warmup: usize, - /// Timed canonicalizations. The reported time is the median. Values below one are treated as - /// one, since a node with no timed run has no time to report. + /// Timed runs. The reported time is the median of them. Values below one are treated as one, + /// since a node with no timed run has no time to report. pub reps: usize, }