vortex_array/aggregate_fn/
session.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4use std::any::Any;
5use std::sync::Arc;
6
7use parking_lot::RwLock;
8use vortex_session::Ref;
9use vortex_session::SessionExt;
10use vortex_session::SessionVar;
11use vortex_session::registry::Registry;
12use vortex_utils::aliases::hash_map::HashMap;
13
14use crate::aggregate_fn::AggregateFnId;
15use crate::aggregate_fn::AggregateFnPluginRef;
16use crate::aggregate_fn::AggregateFnVTable;
17use crate::aggregate_fn::fns::all_nan::AllNan;
18use crate::aggregate_fn::fns::all_non_distinct::AllNonDistinct;
19use crate::aggregate_fn::fns::all_non_nan::AllNonNan;
20use crate::aggregate_fn::fns::all_non_null::AllNonNull;
21use crate::aggregate_fn::fns::all_null::AllNull;
22use crate::aggregate_fn::fns::bounded_max::BoundedMax;
23use crate::aggregate_fn::fns::bounded_min::BoundedMin;
24use crate::aggregate_fn::fns::first::First;
25use crate::aggregate_fn::fns::is_constant::IsConstant;
26use crate::aggregate_fn::fns::is_sorted::IsSorted;
27use crate::aggregate_fn::fns::last::Last;
28use crate::aggregate_fn::fns::max::Max;
29use crate::aggregate_fn::fns::min::Min;
30use crate::aggregate_fn::fns::min_max::MinMax;
31use crate::aggregate_fn::fns::nan_count::NanCount;
32use crate::aggregate_fn::fns::null_count::NullCount;
33use crate::aggregate_fn::fns::sum::Sum;
34use crate::aggregate_fn::fns::uncompressed_size_in_bytes::UncompressedSizeInBytes;
35use crate::aggregate_fn::kernels::DynAggregateKernel;
36use crate::aggregate_fn::kernels::DynGroupedAggregateKernel;
37use crate::array::ArrayId;
38use crate::array::VTable;
39use crate::arrays::Chunked;
40use crate::arrays::Dict;
41use crate::arrays::chunked::compute::aggregate::ChunkedArrayAggregate;
42use crate::arrays::dict::compute::is_constant::DictIsConstantKernel;
43use crate::arrays::dict::compute::is_sorted::DictIsSortedKernel;
44use crate::arrays::dict::compute::min_max::DictMinMaxKernel;
45
46/// Registry of aggregate function vtables.
47pub type AggregateFnRegistry = Registry<AggregateFnPluginRef>;
48
49/// Session state for aggregate function vtables.
50#[derive(Debug)]
51pub struct AggregateFnSession {
52    registry: AggregateFnRegistry,
53
54    pub(super) kernels: RwLock<HashMap<KernelKey, &'static dyn DynAggregateKernel>>,
55    pub(super) grouped_kernels: RwLock<HashMap<KernelKey, &'static dyn DynGroupedAggregateKernel>>,
56}
57
58impl SessionVar for AggregateFnSession {
59    fn as_any(&self) -> &dyn Any {
60        self
61    }
62
63    fn as_any_mut(&mut self) -> &mut dyn Any {
64        self
65    }
66}
67
68type KernelKey = (ArrayId, Option<AggregateFnId>);
69
70impl Default for AggregateFnSession {
71    fn default() -> Self {
72        let this = Self {
73            registry: AggregateFnRegistry::default(),
74            kernels: RwLock::new(HashMap::default()),
75            grouped_kernels: RwLock::new(HashMap::default()),
76        };
77
78        // Register the built-in aggregate functions
79        this.register(AllNonDistinct);
80        this.register(AllNonNan);
81        this.register(AllNonNull);
82        this.register(AllNan);
83        this.register(AllNull);
84        this.register(BoundedMax);
85        this.register(BoundedMin);
86        this.register(First);
87        this.register(IsConstant);
88        this.register(IsSorted);
89        this.register(Last);
90        this.register(Max);
91        this.register(Min);
92        this.register(MinMax);
93        this.register(NanCount);
94        this.register(NullCount);
95        this.register(Sum);
96        this.register(UncompressedSizeInBytes);
97
98        // Register the built-in aggregate kernels.
99        this.register_aggregate_kernel(Chunked.id(), None::<AggregateFnId>, &ChunkedArrayAggregate);
100        this.register_aggregate_kernel(Dict.id(), Some(MinMax.id()), &DictMinMaxKernel);
101        this.register_aggregate_kernel(Dict.id(), Some(IsConstant.id()), &DictIsConstantKernel);
102        this.register_aggregate_kernel(Dict.id(), Some(IsSorted.id()), &DictIsSortedKernel);
103
104        this
105    }
106}
107
108impl AggregateFnSession {
109    /// Returns the aggregate function registry.
110    pub fn registry(&self) -> &AggregateFnRegistry {
111        &self.registry
112    }
113
114    /// Register an aggregate function vtable in the session, replacing any existing vtable with
115    /// the same ID.
116    pub fn register<V: AggregateFnVTable>(&self, vtable: V) {
117        self.registry
118            .register(vtable.id(), Arc::new(vtable) as AggregateFnPluginRef);
119    }
120
121    /// Register an aggregate function kernel for a specific aggregate function and array type.
122    pub fn register_aggregate_kernel(
123        &self,
124        array_id: impl Into<ArrayId>,
125        agg_fn_id: Option<impl Into<AggregateFnId>>,
126        kernel: &'static dyn DynAggregateKernel,
127    ) {
128        self.kernels
129            .write()
130            .insert((array_id.into(), agg_fn_id.map(|id| id.into())), kernel);
131    }
132}
133
134/// Extension trait for accessing aggregate function session data.
135pub trait AggregateFnSessionExt: SessionExt {
136    /// Returns the aggregate function vtable registry.
137    fn aggregate_fns(&self) -> Ref<'_, AggregateFnSession> {
138        self.get::<AggregateFnSession>()
139    }
140}
141impl<S: SessionExt> AggregateFnSessionExt for S {}
vortex_array/aggregate_fn/session.rs

vortex_array/aggregate_fn/
session.rs
