vortex_array/aggregate_fn/fns/all_non_null/
mod.rs1use vortex_error::VortexResult;
5use vortex_session::VortexSession;
6use vortex_session::registry::CachedId;
7
8use crate::ArrayRef;
9use crate::Columnar;
10use crate::ExecutionCtx;
11use crate::IntoArray;
12use crate::aggregate_fn::AggregateArgs;
13use crate::aggregate_fn::AggregateFnId;
14use crate::aggregate_fn::AggregateFnVTable;
15use crate::aggregate_fn::EmptyOptions;
16use crate::dtype::DType;
17use crate::dtype::Nullability;
18use crate::scalar::Scalar;
19
20#[derive(Clone, Debug)]
24pub struct AllNonNull;
25
26impl AggregateFnVTable for AllNonNull {
27 type Options = EmptyOptions;
28 type Partial = bool;
29
30 fn id(&self) -> AggregateFnId {
31 static ID: CachedId = CachedId::new("vortex.all_non_null");
32 *ID
33 }
34
35 fn serialize(&self, _options: &Self::Options) -> VortexResult<Option<Vec<u8>>> {
36 Ok(Some(vec![]))
37 }
38
39 fn deserialize(
40 &self,
41 _metadata: &[u8],
42 _session: &VortexSession,
43 ) -> VortexResult<Self::Options> {
44 Ok(EmptyOptions)
45 }
46
47 fn return_dtype(&self, _options: &Self::Options, _input_dtype: &DType) -> Option<DType> {
48 Some(DType::Bool(Nullability::NonNullable))
49 }
50
51 fn partial_dtype(&self, options: &Self::Options, input_dtype: &DType) -> Option<DType> {
52 self.return_dtype(options, input_dtype)
53 }
54
55 fn empty_partial(
56 &self,
57 _args: AggregateArgs<'_, Self::Options>,
58 ) -> VortexResult<Self::Partial> {
59 Ok(true)
60 }
61
62 fn partial_from_scalar(
63 &self,
64 _args: AggregateArgs<'_, Self::Options>,
65 scalar: &Scalar,
66 ) -> VortexResult<Self::Partial> {
67 bool::try_from(scalar)
68 }
69
70 fn merge_partials(
71 &self,
72 _args: AggregateArgs<'_, Self::Options>,
73 first: Self::Partial,
74 second: Self::Partial,
75 ) -> VortexResult<Self::Partial> {
76 Ok(first && second)
77 }
78
79 fn to_scalar(
80 &self,
81 _args: AggregateArgs<'_, Self::Options>,
82 partial: &Self::Partial,
83 ) -> VortexResult<Scalar> {
84 Ok(Scalar::bool(*partial, Nullability::NonNullable))
85 }
86
87 fn is_saturated(
88 &self,
89 _args: AggregateArgs<'_, Self::Options>,
90 partial: &Self::Partial,
91 ) -> bool {
92 !*partial
93 }
94
95 fn try_accumulate(
96 &self,
97 _args: AggregateArgs<'_, Self::Options>,
98 state: &mut Self::Partial,
99 batch: &ArrayRef,
100 ctx: &mut ExecutionCtx,
101 ) -> VortexResult<bool> {
102 *state &= batch.invalid_count(ctx)? == 0;
103 Ok(true)
104 }
105
106 fn accumulate(
107 &self,
108 _args: AggregateArgs<'_, Self::Options>,
109 partial: &mut Self::Partial,
110 batch: &Columnar,
111 ctx: &mut ExecutionCtx,
112 ) -> VortexResult<()> {
113 *partial &= match batch {
116 Columnar::Constant(c) => c.is_empty() || !c.scalar().is_null(),
117 Columnar::Canonical(c) => c.clone().into_array().invalid_count(ctx)? == 0,
118 };
119 Ok(())
120 }
121
122 fn finalize(
123 &self,
124 _args: AggregateArgs<'_, Self::Options>,
125 partials: ArrayRef,
126 ) -> VortexResult<ArrayRef> {
127 Ok(partials)
128 }
129
130 fn finalize_scalar(
131 &self,
132 args: AggregateArgs<'_, Self::Options>,
133 partial: &Self::Partial,
134 ) -> VortexResult<Scalar> {
135 self.to_scalar(args, partial)
136 }
137}
138
139#[cfg(test)]
140mod tests {
141 use vortex_error::VortexResult;
142
143 use crate::IntoArray;
144 use crate::VortexSessionExecute;
145 use crate::aggregate_fn::Accumulator;
146 use crate::aggregate_fn::DynAccumulator;
147 use crate::aggregate_fn::EmptyOptions;
148 use crate::aggregate_fn::fns::all_non_null::AllNonNull;
149 use crate::array_session;
150 use crate::arrays::PrimitiveArray;
151 use crate::dtype::DType;
152 use crate::dtype::Nullability;
153 use crate::dtype::PType;
154
155 #[test]
156 fn all_non_null_aggregate_fn() -> VortexResult<()> {
157 let mut ctx = array_session().create_execution_ctx();
158 let dtype = DType::Primitive(PType::I32, Nullability::Nullable);
159 let mut acc = Accumulator::try_new(AllNonNull, EmptyOptions, dtype)?;
160
161 let batch = PrimitiveArray::from_option_iter([Some(1i32), Some(2), Some(3)]).into_array();
162 acc.accumulate(&batch, &mut ctx)?;
163
164 assert!(bool::try_from(&acc.finish()?)?);
165 Ok(())
166 }
167
168 #[test]
169 fn all_non_null_false_with_nulls() -> VortexResult<()> {
170 let mut ctx = array_session().create_execution_ctx();
171 let dtype = DType::Primitive(PType::I32, Nullability::Nullable);
172 let mut acc = Accumulator::try_new(AllNonNull, EmptyOptions, dtype)?;
173
174 let batch = PrimitiveArray::from_option_iter([Some(1i32), None, Some(3)]).into_array();
175 acc.accumulate(&batch, &mut ctx)?;
176
177 assert!(!bool::try_from(&acc.finish()?)?);
178 Ok(())
179 }
180
181 #[test]
182 fn all_non_null_true_for_empty_input() -> VortexResult<()> {
183 let mut ctx = array_session().create_execution_ctx();
184 let dtype = DType::Primitive(PType::I32, Nullability::Nullable);
185 let mut acc = Accumulator::try_new(AllNonNull, EmptyOptions, dtype)?;
186
187 let batch = PrimitiveArray::empty::<i32>(Nullability::Nullable).into_array();
188 acc.accumulate(&batch, &mut ctx)?;
189
190 assert!(bool::try_from(&acc.finish()?)?);
191 Ok(())
192 }
193}