vortex_array/aggregate_fn/
session.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4use std::sync::Arc;
5
6use parking_lot::RwLock;
7use vortex_session::Ref;
8use vortex_session::SessionExt;
9use vortex_session::registry::Registry;
10use vortex_utils::aliases::hash_map::HashMap;
11
12use crate::aggregate_fn::AggregateFnId;
13use crate::aggregate_fn::AggregateFnPluginRef;
14use crate::aggregate_fn::AggregateFnVTable;
15use crate::aggregate_fn::fns::is_constant::IsConstant;
16use crate::aggregate_fn::fns::is_sorted::IsSorted;
17use crate::aggregate_fn::fns::min_max::MinMax;
18use crate::aggregate_fn::fns::nan_count::NanCount;
19use crate::aggregate_fn::fns::sum::Sum;
20use crate::aggregate_fn::kernels::DynAggregateKernel;
21use crate::aggregate_fn::kernels::DynGroupedAggregateKernel;
22use crate::arrays::Chunked;
23use crate::arrays::Dict;
24use crate::arrays::chunked::compute::aggregate::ChunkedArrayAggregate;
25use crate::arrays::dict::compute::is_constant::DictIsConstantKernel;
26use crate::arrays::dict::compute::is_sorted::DictIsSortedKernel;
27use crate::arrays::dict::compute::min_max::DictMinMaxKernel;
28use crate::vtable::ArrayId;
29
30/// Registry of aggregate function vtables.
31pub type AggregateFnRegistry = Registry<AggregateFnPluginRef>;
32
33/// Session state for aggregate function vtables.
34#[derive(Debug)]
35pub struct AggregateFnSession {
36    registry: AggregateFnRegistry,
37
38    pub(super) kernels: RwLock<HashMap<KernelKey, &'static dyn DynAggregateKernel>>,
39    pub(super) grouped_kernels: RwLock<HashMap<KernelKey, &'static dyn DynGroupedAggregateKernel>>,
40}
41
42type KernelKey = (ArrayId, Option<AggregateFnId>);
43
44impl Default for AggregateFnSession {
45    fn default() -> Self {
46        let this = Self {
47            registry: AggregateFnRegistry::default(),
48            kernels: RwLock::new(HashMap::default()),
49            grouped_kernels: RwLock::new(HashMap::default()),
50        };
51
52        // Register the built-in aggregate functions
53        this.register(IsConstant);
54        this.register(IsSorted);
55        this.register(MinMax);
56        this.register(NanCount);
57        this.register(Sum);
58
59        // Register the built-in aggregate kernels.
60        this.register_aggregate_kernel(Chunked::ID, None, &ChunkedArrayAggregate);
61        this.register_aggregate_kernel(Dict::ID, Some(MinMax.id()), &DictMinMaxKernel);
62        this.register_aggregate_kernel(Dict::ID, Some(IsConstant.id()), &DictIsConstantKernel);
63        this.register_aggregate_kernel(Dict::ID, Some(IsSorted.id()), &DictIsSortedKernel);
64
65        this
66    }
67}
68
69impl AggregateFnSession {
70    /// Returns the aggregate function registry.
71    pub fn registry(&self) -> &AggregateFnRegistry {
72        &self.registry
73    }
74
75    /// Register an aggregate function vtable in the session, replacing any existing vtable with
76    /// the same ID.
77    pub fn register<V: AggregateFnVTable>(&self, vtable: V) {
78        self.registry
79            .register(vtable.id(), Arc::new(vtable) as AggregateFnPluginRef);
80    }
81
82    /// Register an aggregate function kernel for a specific aggregate function and array type.
83    pub fn register_aggregate_kernel(
84        &self,
85        array_id: ArrayId,
86        agg_fn_id: Option<AggregateFnId>,
87        kernel: &'static dyn DynAggregateKernel,
88    ) {
89        self.kernels.write().insert((array_id, agg_fn_id), kernel);
90    }
91}
92
93/// Extension trait for accessing aggregate function session data.
94pub trait AggregateFnSessionExt: SessionExt {
95    /// Returns the aggregate function vtable registry.
96    fn aggregate_fns(&self) -> Ref<'_, AggregateFnSession> {
97        self.get::<AggregateFnSession>()
98    }
99}
100impl<S: SessionExt> AggregateFnSessionExt for S {}
vortex_array/aggregate_fn/session.rs

vortex_array/aggregate_fn/
session.rs
