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::first::First;
18use crate::aggregate_fn::fns::is_constant::IsConstant;
19use crate::aggregate_fn::fns::is_sorted::IsSorted;
20use crate::aggregate_fn::fns::last::Last;
21use crate::aggregate_fn::fns::min_max::MinMax;
22use crate::aggregate_fn::fns::nan_count::NanCount;
23use crate::aggregate_fn::fns::sum::Sum;
24use crate::aggregate_fn::kernels::DynAggregateKernel;
25use crate::aggregate_fn::kernels::DynGroupedAggregateKernel;
26use crate::array::ArrayId;
27use crate::array::VTable;
28use crate::arrays::Chunked;
29use crate::arrays::Dict;
30use crate::arrays::chunked::compute::aggregate::ChunkedArrayAggregate;
31use crate::arrays::dict::compute::is_constant::DictIsConstantKernel;
32use crate::arrays::dict::compute::is_sorted::DictIsSortedKernel;
33use crate::arrays::dict::compute::min_max::DictMinMaxKernel;
34
35/// Registry of aggregate function vtables.
36pub type AggregateFnRegistry = Registry<AggregateFnPluginRef>;
37
38/// Session state for aggregate function vtables.
39#[derive(Debug)]
40pub struct AggregateFnSession {
41    registry: AggregateFnRegistry,
42
43    pub(super) kernels: RwLock<HashMap<KernelKey, &'static dyn DynAggregateKernel>>,
44    pub(super) grouped_kernels: RwLock<HashMap<KernelKey, &'static dyn DynGroupedAggregateKernel>>,
45}
46
47impl SessionVar for AggregateFnSession {
48    fn as_any(&self) -> &dyn Any {
49        self
50    }
51
52    fn as_any_mut(&mut self) -> &mut dyn Any {
53        self
54    }
55}
56
57type KernelKey = (ArrayId, Option<AggregateFnId>);
58
59impl Default for AggregateFnSession {
60    fn default() -> Self {
61        let this = Self {
62            registry: AggregateFnRegistry::default(),
63            kernels: RwLock::new(HashMap::default()),
64            grouped_kernels: RwLock::new(HashMap::default()),
65        };
66
67        // Register the built-in aggregate functions
68        this.register(First);
69        this.register(IsConstant);
70        this.register(IsSorted);
71        this.register(Last);
72        this.register(MinMax);
73        this.register(NanCount);
74        this.register(Sum);
75
76        // Register the built-in aggregate kernels.
77        this.register_aggregate_kernel(Chunked.id(), None::<AggregateFnId>, &ChunkedArrayAggregate);
78        this.register_aggregate_kernel(Dict.id(), Some(MinMax.id()), &DictMinMaxKernel);
79        this.register_aggregate_kernel(Dict.id(), Some(IsConstant.id()), &DictIsConstantKernel);
80        this.register_aggregate_kernel(Dict.id(), Some(IsSorted.id()), &DictIsSortedKernel);
81
82        this
83    }
84}
85
86impl AggregateFnSession {
87    /// Returns the aggregate function registry.
88    pub fn registry(&self) -> &AggregateFnRegistry {
89        &self.registry
90    }
91
92    /// Register an aggregate function vtable in the session, replacing any existing vtable with
93    /// the same ID.
94    pub fn register<V: AggregateFnVTable>(&self, vtable: V) {
95        self.registry
96            .register(vtable.id(), Arc::new(vtable) as AggregateFnPluginRef);
97    }
98
99    /// Register an aggregate function kernel for a specific aggregate function and array type.
100    pub fn register_aggregate_kernel(
101        &self,
102        array_id: impl Into<ArrayId>,
103        agg_fn_id: Option<impl Into<AggregateFnId>>,
104        kernel: &'static dyn DynAggregateKernel,
105    ) {
106        self.kernels
107            .write()
108            .insert((array_id.into(), agg_fn_id.map(|id| id.into())), kernel);
109    }
110}
111
112/// Extension trait for accessing aggregate function session data.
113pub trait AggregateFnSessionExt: SessionExt {
114    /// Returns the aggregate function vtable registry.
115    fn aggregate_fns(&self) -> Ref<'_, AggregateFnSession> {
116        self.get::<AggregateFnSession>()
117    }
118}
119impl<S: SessionExt> AggregateFnSessionExt for S {}
vortex_array/aggregate_fn/session.rs

vortex_array/aggregate_fn/
session.rs
