Skip to content

Commit 802a199

Browse files
robert3005claude
andauthored
Speed up scalar construction, stats reads and aggregate partial merging (#10326)
## Summary Make stats validation cheap, skips it where values are valid by construction, and removes `Scalar` construction and copies from stats reads and aggregate partial merging. ## Changes - `Scalar::try_new` and `Scalar::validate` run an inline fast check. Error messages are built in a cold path. - Constructors that are valid by construction use `new_unchecked`. VarBinView reads no longer check UTF-8 again. - Stats reads and merges build fewer scalars. Cached bool and count stats are read without a `Scalar`. - SumV2 parses partials from the tuple values directly. - Accumulators made from one `AggregateFnRef` share their resolved dtypes, so their partial dtypes compare by pointer. - Fix: `merge_is_constant` always overwrote its result with `exact(false)`. - New `scalar_construction` bench. | ns/op (M4 Max) | Before | After | |---|---|---| | `try_new`, primitive | 8.6 | 1.3 | | `execute_scalar`, primitive / utf8 / struct8 | 15.9 / 26.2 / 196 | 7.5 / 11.8 / 114 | | `statistics().get(Min)` | 18.0 | 9.7 | | `StatsSet::merge_ordered` | 704 | 300 | | SumV2 `combine_partials` | 99 | 31 | ## API Changes - `AggregateFnVTable::partial_from_scalar`, `DynAccumulator::combine_partials` and `MinMaxResult::from_scalar` take `&Scalar`. - New `StructFields::field_dtypes()` returns `&DType` without cloning. - After a stats merge, `IsConstant` can now be `exact(true)`. When the result cannot be known, the stat is cleared. --------- Signed-off-by: Robert Kruszewski <github@robertk.io> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 5314209 commit 802a199

46 files changed

Lines changed: 943 additions & 236 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎encodings/runend/src/compute/sum/tests.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -54,7 +54,7 @@ fn check_sum(array: ArrayRef, options: NumericalAggregateOpts) -> VortexResult<(
5454
.aggregate(&aggregate, &array, &mut ctx)?
5555
.vortex_expect("The fixture has a primitive dtype supported by the run-end sum kernel");
5656
let mut direct = aggregate.accumulator(array.dtype())?;
57-
direct.combine_partials(partial)?;
57+
direct.combine_partials(&partial)?;
5858

5959
let mut dispatched = aggregate.accumulator(array.dtype())?;
6060
dispatched.accumulate(&array, &mut ctx)?;

‎encodings/sparse/src/compute/sum.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -125,7 +125,7 @@ mod tests {
125125
};
126126

127127
let mut accumulator = Accumulator::try_new(SumV2, options, arr.dtype().clone())?;
128-
accumulator.combine_partials(partial)?;
128+
accumulator.combine_partials(&partial)?;
129129
accumulator.finish()
130130
}
131131

‎vortex-array/Cargo.toml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -182,6 +182,10 @@ harness = false
182182
name = "scalar_at_struct"
183183
harness = false
184184

185+
[[bench]]
186+
name = "scalar_construction"
187+
harness = false
188+
185189
[[bench]]
186190
name = "patches_lookup"
187191
harness = false
Lines changed: 211 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,211 @@
1+
// SPDX-License-Identifier: Apache-2.0
2+
// SPDX-FileCopyrightText: Copyright the Vortex contributors
3+
4+
#![expect(clippy::unwrap_used)]
5+
6+
//! Scalar construction and validation, stats reads and merges, and accumulator creation.
7+
//!
8+
//! Every benchmark does `N` operations per iteration, except the stats merge, which does `MERGES`.
9+
10+
use std::hint::black_box;
11+
use std::sync::LazyLock;
12+
13+
use divan::Bencher;
14+
use divan::counter::ItemsCount;
15+
use mimalloc::MiMalloc;
16+
use vortex_array::IntoArray;
17+
use vortex_array::VortexSessionExecute;
18+
use vortex_array::aggregate_fn::AggregateFnVTableExt;
19+
use vortex_array::aggregate_fn::NumericalAggregateOpts;
20+
use vortex_array::aggregate_fn::fns::count::Count;
21+
use vortex_array::aggregate_fn::fns::sum_v2::SumV2;
22+
use vortex_array::array_session;
23+
use vortex_array::dtype::DType;
24+
use vortex_array::dtype::FieldNames;
25+
use vortex_array::dtype::Nullability;
26+
use vortex_array::dtype::PType;
27+
use vortex_array::dtype::StructFields;
28+
use vortex_array::expr::stats::Precision;
29+
use vortex_array::expr::stats::Stat;
30+
use vortex_array::expr::stats::StatsProvider;
31+
use vortex_array::scalar::PValue;
32+
use vortex_array::scalar::Scalar;
33+
use vortex_array::scalar::ScalarValue;
34+
use vortex_array::stats::StatsSet;
35+
use vortex_buffer::Buffer;
36+
use vortex_buffer::BufferString;
37+
use vortex_session::VortexSession;
38+
39+
#[global_allocator]
40+
static GLOBAL: MiMalloc = MiMalloc;
41+
42+
const N: usize = 1024;
43+
/// A merge costs about 300ns, so it runs fewer times to keep the CodSpeed simulation under 1ms.
44+
const MERGES: usize = 64;
45+
46+
static SESSION: LazyLock<VortexSession> = LazyLock::new(array_session);
47+
48+
fn main() {
49+
LazyLock::force(&SESSION);
50+
divan::main();
51+
}
52+
53+
fn struct8_dtype() -> DType {
54+
DType::Struct(
55+
StructFields::new(
56+
FieldNames::from(["f1", "f2", "f3", "f4", "f5", "f6", "f7", "f8"]),
57+
vec![DType::Primitive(PType::I64, Nullability::NonNullable); 8],
58+
),
59+
Nullability::NonNullable,
60+
)
61+
}
62+
63+
fn struct8_value(i: i64) -> ScalarValue {
64+
ScalarValue::Tuple(
65+
(0..8)
66+
.map(|f| Some(ScalarValue::Primitive(PValue::I64(i + f))))
67+
.collect(),
68+
)
69+
}
70+
71+
#[divan::bench]
72+
fn try_new_primitive(bencher: Bencher) {
73+
bencher.counter(ItemsCount::new(N)).bench_local(|| {
74+
for i in 0..N {
75+
black_box(
76+
Scalar::try_new(
77+
DType::Primitive(PType::I64, Nullability::NonNullable),
78+
Some(ScalarValue::Primitive(PValue::I64(black_box(i as i64)))),
79+
)
80+
.unwrap(),
81+
);
82+
}
83+
});
84+
}
85+
86+
#[divan::bench]
87+
fn try_new_null(bencher: Bencher) {
88+
bencher.counter(ItemsCount::new(N)).bench_local(|| {
89+
for _ in 0..N {
90+
black_box(
91+
Scalar::try_new(black_box(DType::Utf8(Nullability::Nullable)), None).unwrap(),
92+
);
93+
}
94+
});
95+
}
96+
97+
#[divan::bench]
98+
fn try_new_utf8(bencher: Bencher) {
99+
let value = BufferString::from("hello world, a string");
100+
bencher.counter(ItemsCount::new(N)).bench_local(|| {
101+
for _ in 0..N {
102+
black_box(
103+
Scalar::try_new(
104+
DType::Utf8(Nullability::NonNullable),
105+
Some(ScalarValue::Utf8(black_box(&value).clone())),
106+
)
107+
.unwrap(),
108+
);
109+
}
110+
});
111+
}
112+
113+
#[divan::bench]
114+
fn try_new_struct8(bencher: Bencher) {
115+
let dtype = struct8_dtype();
116+
bencher.counter(ItemsCount::new(N)).bench_local(|| {
117+
for i in 0..N {
118+
black_box(Scalar::try_new(dtype.clone(), Some(struct8_value(i as i64))).unwrap());
119+
}
120+
});
121+
}
122+
123+
#[divan::bench]
124+
fn validate_struct8(bencher: Bencher) {
125+
let dtype = struct8_dtype();
126+
let value = struct8_value(0);
127+
bencher.counter(ItemsCount::new(N)).bench_local(|| {
128+
for _ in 0..N {
129+
black_box(Scalar::validate(black_box(&dtype), Some(black_box(&value)))).unwrap();
130+
}
131+
});
132+
}
133+
134+
fn i64_stats(offset: i64) -> StatsSet {
135+
StatsSet::from_iter([
136+
(Stat::Min, Precision::exact(ScalarValue::from(offset))),
137+
(Stat::Max, Precision::exact(ScalarValue::from(offset + 100))),
138+
(Stat::Sum, Precision::exact(ScalarValue::from(offset * 7))),
139+
(Stat::NullCount, Precision::exact(ScalarValue::from(3u64))),
140+
(Stat::IsConstant, Precision::exact(ScalarValue::from(false))),
141+
(Stat::IsSorted, Precision::exact(ScalarValue::from(true))),
142+
(
143+
Stat::IsStrictSorted,
144+
Precision::exact(ScalarValue::from(true)),
145+
),
146+
(
147+
Stat::UncompressedSizeInBytes,
148+
Precision::exact(ScalarValue::from(800u64)),
149+
),
150+
])
151+
}
152+
153+
#[divan::bench]
154+
fn stats_get_min(bencher: Bencher) {
155+
let array = (0..1024i64).collect::<Buffer<i64>>().into_array();
156+
array.statistics().set_iter(i64_stats(0).into_iter());
157+
bencher.counter(ItemsCount::new(N)).bench_local(|| {
158+
for _ in 0..N {
159+
black_box(black_box(&array).statistics().get(Stat::Min));
160+
}
161+
});
162+
}
163+
164+
#[divan::bench]
165+
fn stats_compute_is_constant_cached(bencher: Bencher) {
166+
let array = (0..1024i64).collect::<Buffer<i64>>().into_array();
167+
array.statistics().set_iter(i64_stats(0).into_iter());
168+
let mut ctx = SESSION.create_execution_ctx();
169+
bencher.counter(ItemsCount::new(N)).bench_local(|| {
170+
for _ in 0..N {
171+
black_box(black_box(&array).statistics().compute_is_constant(&mut ctx));
172+
}
173+
});
174+
}
175+
176+
#[divan::bench]
177+
fn stats_merge_ordered(bencher: Bencher) {
178+
let dtype = DType::Primitive(PType::I64, Nullability::NonNullable);
179+
let left = i64_stats(0);
180+
let right = i64_stats(1000);
181+
bencher
182+
.counter(ItemsCount::new(MERGES))
183+
.with_inputs(|| vec![left.clone(); MERGES])
184+
.bench_local_values(|sets| {
185+
for set in sets {
186+
black_box(set.merge_ordered(&right, &dtype));
187+
}
188+
});
189+
}
190+
191+
#[divan::bench]
192+
fn accumulator_sum_v2(bencher: Bencher) {
193+
let dtype = DType::Primitive(PType::I64, Nullability::Nullable);
194+
let aggregate_fn = SumV2.bind(NumericalAggregateOpts::default());
195+
bencher.counter(ItemsCount::new(N)).bench_local(|| {
196+
for _ in 0..N {
197+
black_box(aggregate_fn.accumulator(black_box(&dtype)).unwrap());
198+
}
199+
});
200+
}
201+
202+
#[divan::bench]
203+
fn accumulator_count_struct8(bencher: Bencher) {
204+
let dtype = struct8_dtype();
205+
let aggregate_fn = Count.bind(NumericalAggregateOpts::default());
206+
bencher.counter(ItemsCount::new(N)).bench_local(|| {
207+
for _ in 0..N {
208+
black_box(aggregate_fn.accumulator(black_box(&dtype)).unwrap());
209+
}
210+
});
211+
}

‎vortex-array/src/aggregate_fn/accumulator.rs‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -89,7 +89,7 @@ impl<V: AggregateFnVTable> Accumulator<V> {
8989
Ok(())
9090
}
9191

92-
fn fold_partial_scalar(&mut self, scalar: Scalar) -> VortexResult<()> {
92+
fn fold_partial_scalar(&mut self, scalar: &Scalar) -> VortexResult<()> {
9393
let other = self
9494
.vtable
9595
.partial_from_scalar(self.dtypes.args(&self.options), scalar)?;
@@ -113,7 +113,7 @@ pub trait DynAccumulator: 'static + Send {
113113
///
114114
/// The scalar must have the dtype reported by the vtable's `partial_dtype`, and represents
115115
/// input following the input already accumulated.
116-
fn combine_partials(&mut self, partial: Scalar) -> VortexResult<()>;
116+
fn combine_partials(&mut self, partial: &Scalar) -> VortexResult<()>;
117117

118118
/// Whether the accumulator's result is fully determined.
119119
fn is_saturated(&self) -> bool;
@@ -178,7 +178,7 @@ impl<V: AggregateFnVTable> DynAccumulator for Accumulator<V> {
178178
);
179179
partial.cast(&self.dtypes.partial_dtype)?
180180
};
181-
self.fold_partial_scalar(partial)?;
181+
self.fold_partial_scalar(&partial)?;
182182
return Ok(());
183183
}
184184

@@ -201,7 +201,7 @@ impl<V: AggregateFnVTable> DynAccumulator for Accumulator<V> {
201201
&self.dtypes.partial_dtype,
202202
"Aggregate kernel returned the wrong partial dtype",
203203
);
204-
self.fold_partial_scalar(result)?;
204+
self.fold_partial_scalar(&result)?;
205205
return Ok(());
206206
}
207207
}
@@ -237,7 +237,7 @@ impl<V: AggregateFnVTable> DynAccumulator for Accumulator<V> {
237237
&self.dtypes.partial_dtype,
238238
"Aggregate kernel returned the wrong partial dtype",
239239
);
240-
self.fold_partial_scalar(result)?;
240+
self.fold_partial_scalar(&result)?;
241241
return Ok(());
242242
}
243243

@@ -271,7 +271,7 @@ impl<V: AggregateFnVTable> DynAccumulator for Accumulator<V> {
271271
}
272272
}
273273

274-
fn combine_partials(&mut self, partial: Scalar) -> VortexResult<()> {
274+
fn combine_partials(&mut self, partial: &Scalar) -> VortexResult<()> {
275275
vortex_ensure_eq!(
276276
partial.dtype(),
277277
&self.dtypes.partial_dtype,

‎vortex-array/src/aggregate_fn/accumulator_grouped.rs‎

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -195,16 +195,22 @@ pub struct GroupedAccumulator<V: AggregateFnVTable> {
195195

196196
impl<V: AggregateFnVTable> GroupedAccumulator<V> {
197197
pub fn try_new(vtable: V, options: V::Options, dtype: DType) -> VortexResult<Self> {
198-
let aggregate_fn = AggregateFn::new(vtable.clone(), options.clone()).erased();
199198
let dtypes = AggregateDTypes::try_new(&vtable, &options, dtype)?;
200199

201-
Ok(Self {
200+
Ok(Self::from_dtypes(vtable, options, dtypes))
201+
}
202+
203+
/// Build a grouped accumulator over dtypes that are already resolved.
204+
pub(super) fn from_dtypes(vtable: V, options: V::Options, dtypes: AggregateDTypes) -> Self {
205+
let aggregate_fn = AggregateFn::new(vtable.clone(), options.clone()).erased();
206+
207+
Self {
202208
vtable,
203209
options,
204210
aggregate_fn,
205211
dtypes,
206212
partials: vec![],
207-
})
213+
}
208214
}
209215
}
210216

‎vortex-array/src/aggregate_fn/combined.rs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -246,7 +246,7 @@ impl<T: BinaryCombined> AggregateFnVTable for Combined<T> {
246246
fn partial_from_scalar(
247247
&self,
248248
args: AggregateArgs<'_, Self::Options>,
249-
scalar: Scalar,
249+
scalar: &Scalar,
250250
) -> VortexResult<Self::Partial> {
251251
let (mut left, mut right) = self.new_child_accumulators(args)?;
252252
// A null partial represents an empty group and parses to empty child accumulators.
@@ -260,8 +260,8 @@ impl<T: BinaryCombined> AggregateFnVTable for Combined<T> {
260260
let r_field = s
261261
.field(rname)
262262
.ok_or_else(|| vortex_err!("BinaryCombined partial missing `{}` field", rname))?;
263-
left.combine_partials(l_field)?;
264-
right.combine_partials(r_field)?;
263+
left.combine_partials(&l_field)?;
264+
right.combine_partials(&r_field)?;
265265
}
266266
Ok((left, right))
267267
}

‎vortex-array/src/aggregate_fn/fns/all_nan/mod.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -71,9 +71,9 @@ impl AggregateFnVTable for AllNan {
7171
fn partial_from_scalar(
7272
&self,
7373
_args: AggregateArgs<'_, Self::Options>,
74-
scalar: Scalar,
74+
scalar: &Scalar,
7575
) -> VortexResult<Self::Partial> {
76-
bool::try_from(&scalar)
76+
bool::try_from(scalar)
7777
}
7878

7979
fn merge_partials(

‎vortex-array/src/aggregate_fn/fns/all_non_distinct/mod.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -160,7 +160,7 @@ impl AggregateFnVTable for AllNonDistinct {
160160
fn partial_from_scalar(
161161
&self,
162162
_args: AggregateArgs<'_, Self::Options>,
163-
scalar: Scalar,
163+
scalar: &Scalar,
164164
) -> VortexResult<Self::Partial> {
165165
Ok(AllNonDistinctPartial {
166166
all_non_distinct: scalar.as_bool().value().unwrap_or(false),

‎vortex-array/src/aggregate_fn/fns/all_non_nan/mod.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -71,9 +71,9 @@ impl AggregateFnVTable for AllNonNan {
7171
fn partial_from_scalar(
7272
&self,
7373
_args: AggregateArgs<'_, Self::Options>,
74-
scalar: Scalar,
74+
scalar: &Scalar,
7575
) -> VortexResult<Self::Partial> {
76-
bool::try_from(&scalar)
76+
bool::try_from(scalar)
7777
}
7878

7979
fn merge_partials(

0 commit comments

Comments
 (0)