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

vortex_array/aggregate_fn/
session.rs
