1use std::sync::Arc;
7
8use parking_lot::RwLock;
9use vortex_array::ExecutionCtx;
10use vortex_error::VortexError;
11use vortex_error::VortexResult;
12use vortex_error::vortex_panic;
13
14use super::MutTypedStatsSetRef;
15use super::StatsSet;
16use super::StatsSetIntoIter;
17use super::TypedStatsSetRef;
18use crate::ArrayRef;
19use crate::aggregate_fn::fns::is_constant::is_constant;
20use crate::aggregate_fn::fns::is_sorted::is_sorted;
21use crate::aggregate_fn::fns::is_sorted::is_strict_sorted;
22use crate::aggregate_fn::fns::min_max::MinMaxResult;
23use crate::aggregate_fn::fns::min_max::min_max;
24use crate::aggregate_fn::fns::nan_count::nan_count;
25use crate::aggregate_fn::fns::sum::sum;
26use crate::aggregate_fn::fns::uncompressed_size_in_bytes::uncompressed_size_in_bytes;
27use crate::expr::stats::Precision;
28use crate::expr::stats::Stat;
29use crate::expr::stats::StatsProvider;
30use crate::scalar::Scalar;
31use crate::scalar::ScalarValue;
32
33#[derive(Clone, Default, Debug)]
36pub struct ArrayStats {
37 inner: Arc<RwLock<StatsSet>>,
38}
39
40pub struct StatsSetRef<'a> {
44 dyn_array_ref: &'a ArrayRef,
46 array_stats: &'a ArrayStats,
47}
48
49impl ArrayStats {
50 pub fn to_ref<'a>(&'a self, array: &'a ArrayRef) -> StatsSetRef<'a> {
51 StatsSetRef {
52 dyn_array_ref: array,
53 array_stats: self,
54 }
55 }
56
57 pub fn set(&self, stat: Stat, value: Precision<ScalarValue>) {
58 self.inner.write().set(stat, value);
59 }
60
61 pub fn clear(&self, stat: Stat) {
62 self.inner.write().clear(stat);
63 }
64
65 pub fn retain(&self, stats: &[Stat]) {
66 self.inner.write().retain_only(stats);
67 }
68}
69
70impl From<StatsSet> for ArrayStats {
71 fn from(value: StatsSet) -> Self {
72 Self {
73 inner: Arc::new(RwLock::new(value)),
74 }
75 }
76}
77
78impl From<ArrayStats> for StatsSet {
79 fn from(value: ArrayStats) -> Self {
80 value.inner.read().clone()
81 }
82}
83
84impl StatsSetRef<'_> {
85 pub(crate) fn replace(&self, stats: StatsSet) {
86 *self.array_stats.inner.write() = stats;
87 }
88
89 pub fn set_iter(&self, iter: StatsSetIntoIter) {
90 let mut guard = self.array_stats.inner.write();
91 for (stat, value) in iter {
92 guard.set(stat, value);
93 }
94 }
95
96 pub fn inherit_from(&self, stats: StatsSetRef<'_>) {
97 if !Arc::ptr_eq(&self.array_stats.inner, &stats.array_stats.inner) {
99 stats.with_iter(|iter| self.inherit(iter));
100 }
101 }
102
103 pub fn inherit<'a>(&self, iter: impl Iterator<Item = &'a (Stat, Precision<ScalarValue>)>) {
104 let mut guard = self.array_stats.inner.write();
105 for (stat, value) in iter {
106 if !value.is_exact() {
107 if !guard.get(*stat).is_some_and(|v| v.is_exact()) {
108 guard.set(*stat, value.clone());
109 }
110 } else {
111 guard.set(*stat, value.clone());
112 }
113 }
114 }
115
116 pub fn with_typed_stats_set<U, F: FnOnce(TypedStatsSetRef) -> U>(&self, apply: F) -> U {
117 apply(
118 self.array_stats
119 .inner
120 .read()
121 .as_typed_ref(self.dyn_array_ref.dtype()),
122 )
123 }
124
125 pub fn with_mut_typed_stats_set<U, F: FnOnce(MutTypedStatsSetRef) -> U>(&self, apply: F) -> U {
126 apply(
127 self.array_stats
128 .inner
129 .write()
130 .as_mut_typed_ref(self.dyn_array_ref.dtype()),
131 )
132 }
133
134 pub fn to_owned(&self) -> StatsSet {
135 self.array_stats.inner.read().clone()
136 }
137
138 pub fn to_array_stats(&self) -> ArrayStats {
142 self.array_stats.clone()
143 }
144
145 pub fn with_iter<
146 F: for<'a> FnOnce(&mut dyn Iterator<Item = &'a (Stat, Precision<ScalarValue>)>) -> R,
147 R,
148 >(
149 &self,
150 f: F,
151 ) -> R {
152 let lock = self.array_stats.inner.read();
153 f(&mut lock.iter())
154 }
155
156 pub fn compute_stat(&self, stat: Stat, ctx: &mut ExecutionCtx) -> VortexResult<Option<Scalar>> {
157 if let Some(Precision::Exact(s)) = self.get(stat) {
159 return Ok(Some(s));
160 }
161
162 Ok(match stat {
163 Stat::Min => min_max(self.dyn_array_ref, ctx)?.map(|MinMaxResult { min, max: _ }| min),
164 Stat::Max => min_max(self.dyn_array_ref, ctx)?.map(|MinMaxResult { min: _, max }| max),
165 Stat::Sum => {
166 Stat::Sum
167 .dtype(self.dyn_array_ref.dtype())
168 .is_some()
169 .then(|| {
170 sum(self.dyn_array_ref, ctx)
172 })
173 .transpose()?
174 }
175 Stat::NullCount => self.dyn_array_ref.invalid_count(ctx).ok().map(Into::into),
176 Stat::IsConstant => {
177 if self.dyn_array_ref.is_empty() {
178 None
179 } else {
180 Some(is_constant(self.dyn_array_ref, ctx)?.into())
181 }
182 }
183 Stat::IsSorted => Some(is_sorted(self.dyn_array_ref, ctx)?.into()),
184 Stat::IsStrictSorted => Some(is_strict_sorted(self.dyn_array_ref, ctx)?.into()),
185 Stat::UncompressedSizeInBytes => Stat::UncompressedSizeInBytes
186 .dtype(self.dyn_array_ref.dtype())
187 .is_some()
188 .then(|| uncompressed_size_in_bytes(self.dyn_array_ref, ctx))
189 .transpose()?
190 .map(|s| s.into()),
191 Stat::NaNCount => {
192 Stat::NaNCount
193 .dtype(self.dyn_array_ref.dtype())
194 .is_some()
195 .then(|| {
196 nan_count(self.dyn_array_ref, ctx)
198 })
199 .transpose()?
200 .map(|s| s.into())
201 }
202 })
203 }
204
205 pub fn compute_all(&self, stats: &[Stat], ctx: &mut ExecutionCtx) -> VortexResult<StatsSet> {
206 let mut stats_set = StatsSet::default();
207 for &stat in stats {
208 if let Some(s) = self.compute_stat(stat, ctx)?
209 && let Some(value) = s.into_value()
210 {
211 stats_set.set(stat, Precision::exact(value));
212 }
213 }
214 Ok(stats_set)
215 }
216}
217
218impl StatsSetRef<'_> {
219 pub fn compute_as<U: for<'a> TryFrom<&'a Scalar, Error = VortexError>>(
220 &self,
221 stat: Stat,
222 ctx: &mut ExecutionCtx,
223 ) -> Option<U> {
224 self.compute_stat(stat, ctx)
225 .inspect_err(|e| tracing::warn!("Failed to compute stat {stat}: {e}"))
226 .ok()
227 .flatten()
228 .map(|s| U::try_from(&s))
229 .transpose()
230 .unwrap_or_else(|err| {
231 vortex_panic!(
232 err,
233 "Failed to compute stat {} as {}",
234 stat,
235 std::any::type_name::<U>()
236 )
237 })
238 }
239
240 pub fn set(&self, stat: Stat, value: Precision<ScalarValue>) {
241 self.array_stats.set(stat, value);
242 }
243
244 pub fn clear(&self, stat: Stat) {
245 self.array_stats.clear(stat);
246 }
247
248 pub fn compute_min<U: for<'a> TryFrom<&'a Scalar, Error = VortexError>>(
249 &self,
250 ctx: &mut ExecutionCtx,
251 ) -> Option<U> {
252 self.compute_as(Stat::Min, ctx)
253 }
254
255 pub fn compute_max<U: for<'a> TryFrom<&'a Scalar, Error = VortexError>>(
256 &self,
257 ctx: &mut ExecutionCtx,
258 ) -> Option<U> {
259 self.compute_as(Stat::Max, ctx)
260 }
261
262 pub fn compute_is_sorted(&self, ctx: &mut ExecutionCtx) -> Option<bool> {
263 self.compute_as(Stat::IsSorted, ctx)
264 }
265
266 pub fn compute_is_strict_sorted(&self, ctx: &mut ExecutionCtx) -> Option<bool> {
267 self.compute_as(Stat::IsStrictSorted, ctx)
268 }
269
270 pub fn compute_is_constant(&self, ctx: &mut ExecutionCtx) -> Option<bool> {
271 self.compute_as(Stat::IsConstant, ctx)
272 }
273
274 pub fn compute_null_count(&self, ctx: &mut ExecutionCtx) -> Option<usize> {
275 self.compute_as(Stat::NullCount, ctx)
276 }
277
278 pub fn compute_uncompressed_size_in_bytes(&self, ctx: &mut ExecutionCtx) -> Option<usize> {
279 self.compute_as(Stat::UncompressedSizeInBytes, ctx)
280 }
281}
282
283impl StatsProvider for StatsSetRef<'_> {
284 fn get(&self, stat: Stat) -> Option<Precision<Scalar>> {
285 self.array_stats
286 .inner
287 .read()
288 .as_typed_ref(self.dyn_array_ref.dtype())
289 .get(stat)
290 }
291
292 fn len(&self) -> usize {
293 self.array_stats.inner.read().len()
294 }
295}