datafusion_functions_aggregate/
bool_and_or.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! Defines physical expressions that can evaluated at runtime during query execution
19
20use std::any::Any;
21use std::mem::size_of_val;
22
23use arrow::array::ArrayRef;
24use arrow::array::BooleanArray;
25use arrow::compute::bool_and as compute_bool_and;
26use arrow::compute::bool_or as compute_bool_or;
27use arrow::datatypes::Field;
28use arrow::datatypes::{DataType, FieldRef};
29
30use datafusion_common::internal_err;
31use datafusion_common::{Result, ScalarValue};
32use datafusion_common::{downcast_value, not_impl_err};
33use datafusion_expr::function::{AccumulatorArgs, StateFieldsArgs};
34use datafusion_expr::utils::{AggregateOrderSensitivity, format_state_name};
35use datafusion_expr::{
36    Accumulator, AggregateUDFImpl, Documentation, GroupsAccumulator, ReversedUDAF,
37    Signature, Volatility,
38};
39
40use datafusion_functions_aggregate_common::aggregate::groups_accumulator::bool_op::BooleanGroupsAccumulator;
41use datafusion_macros::user_doc;
42
43// returns the new value after bool_and/bool_or with the new values, taking nullability into account
44macro_rules! typed_bool_and_or_batch {
45    ($VALUES:expr, $ARRAYTYPE:ident, $SCALAR:ident, $OP:ident) => {{
46        let array = downcast_value!($VALUES, $ARRAYTYPE);
47        let delta = $OP(array);
48        Ok(ScalarValue::$SCALAR(delta))
49    }};
50}
51
52// bool_and/bool_or the array and returns a ScalarValue of its corresponding type.
53macro_rules! bool_and_or_batch {
54    ($VALUES:expr, $OP:ident) => {{
55        match $VALUES.data_type() {
56            DataType::Boolean => {
57                typed_bool_and_or_batch!($VALUES, BooleanArray, Boolean, $OP)
58            }
59            e => {
60                return internal_err!(
61                    "Bool and/Bool or is not expected to receive the type {e:?}"
62                );
63            }
64        }
65    }};
66}
67
68/// dynamically-typed bool_and(array) -> ScalarValue
69fn bool_and_batch(values: &ArrayRef) -> Result<ScalarValue> {
70    bool_and_or_batch!(values, compute_bool_and)
71}
72
73/// dynamically-typed bool_or(array) -> ScalarValue
74fn bool_or_batch(values: &ArrayRef) -> Result<ScalarValue> {
75    bool_and_or_batch!(values, compute_bool_or)
76}
77
78make_udaf_expr_and_func!(
79    BoolAnd,
80    bool_and,
81    expression,
82    "The values to combine with `AND`",
83    bool_and_udaf
84);
85
86make_udaf_expr_and_func!(
87    BoolOr,
88    bool_or,
89    expression,
90    "The values to combine with `OR`",
91    bool_or_udaf
92);
93
94#[user_doc(
95    doc_section(label = "General Functions"),
96    description = "Returns true if all non-null input values are true, otherwise false.",
97    syntax_example = "bool_and(expression)",
98    sql_example = r#"```sql
99> SELECT bool_and(column_name) FROM table_name;
100+----------------------------+
101| bool_and(column_name)       |
102+----------------------------+
103| true                        |
104+----------------------------+
105```"#,
106    standard_argument(name = "expression", prefix = "The")
107)]
108/// BOOL_AND aggregate expression
109#[derive(Debug, PartialEq, Eq, Hash)]
110pub struct BoolAnd {
111    signature: Signature,
112}
113
114impl BoolAnd {
115    fn new() -> Self {
116        Self {
117            signature: Signature::uniform(
118                1,
119                vec![DataType::Boolean],
120                Volatility::Immutable,
121            ),
122        }
123    }
124}
125
126impl Default for BoolAnd {
127    fn default() -> Self {
128        Self::new()
129    }
130}
131
132impl AggregateUDFImpl for BoolAnd {
133    fn as_any(&self) -> &dyn Any {
134        self
135    }
136
137    fn name(&self) -> &str {
138        "bool_and"
139    }
140
141    fn signature(&self) -> &Signature {
142        &self.signature
143    }
144
145    fn return_type(&self, _: &[DataType]) -> Result<DataType> {
146        Ok(DataType::Boolean)
147    }
148
149    fn accumulator(&self, _: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
150        Ok(Box::<BoolAndAccumulator>::default())
151    }
152
153    fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
154        Ok(vec![
155            Field::new(
156                format_state_name(args.name, self.name()),
157                DataType::Boolean,
158                true,
159            )
160            .into(),
161        ])
162    }
163
164    fn groups_accumulator_supported(&self, _args: AccumulatorArgs) -> bool {
165        true
166    }
167
168    fn create_groups_accumulator(
169        &self,
170        args: AccumulatorArgs,
171    ) -> Result<Box<dyn GroupsAccumulator>> {
172        match args.return_field.data_type() {
173            DataType::Boolean => {
174                Ok(Box::new(BooleanGroupsAccumulator::new(|x, y| x && y, true)))
175            }
176            _ => not_impl_err!(
177                "GroupsAccumulator not supported for {} with {}",
178                args.name,
179                args.return_field.data_type()
180            ),
181        }
182    }
183
184    fn order_sensitivity(&self) -> AggregateOrderSensitivity {
185        AggregateOrderSensitivity::Insensitive
186    }
187
188    fn reverse_expr(&self) -> ReversedUDAF {
189        ReversedUDAF::Identical
190    }
191
192    fn documentation(&self) -> Option<&Documentation> {
193        self.doc()
194    }
195}
196
197#[derive(Debug, Default)]
198struct BoolAndAccumulator {
199    acc: Option<bool>,
200}
201
202impl Accumulator for BoolAndAccumulator {
203    fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
204        let values = &values[0];
205        self.acc = match (self.acc, bool_and_batch(values)?) {
206            (None, ScalarValue::Boolean(v)) => v,
207            (Some(v), ScalarValue::Boolean(None)) => Some(v),
208            (Some(a), ScalarValue::Boolean(Some(b))) => Some(a && b),
209            _ => unreachable!(),
210        };
211        Ok(())
212    }
213
214    fn evaluate(&mut self) -> Result<ScalarValue> {
215        Ok(ScalarValue::Boolean(self.acc))
216    }
217
218    fn size(&self) -> usize {
219        size_of_val(self)
220    }
221
222    fn state(&mut self) -> Result<Vec<ScalarValue>> {
223        Ok(vec![ScalarValue::Boolean(self.acc)])
224    }
225
226    fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
227        self.update_batch(states)
228    }
229}
230
231#[user_doc(
232    doc_section(label = "General Functions"),
233    description = "Returns true if all non-null input values are true, otherwise false.",
234    syntax_example = "bool_and(expression)",
235    sql_example = r#"```sql
236> SELECT bool_and(column_name) FROM table_name;
237+----------------------------+
238| bool_and(column_name)       |
239+----------------------------+
240| true                        |
241+----------------------------+
242```"#,
243    standard_argument(name = "expression", prefix = "The")
244)]
245/// BOOL_OR aggregate expression
246#[derive(Debug, Clone, PartialEq, Eq, Hash)]
247pub struct BoolOr {
248    signature: Signature,
249}
250
251impl BoolOr {
252    fn new() -> Self {
253        Self {
254            signature: Signature::uniform(
255                1,
256                vec![DataType::Boolean],
257                Volatility::Immutable,
258            ),
259        }
260    }
261}
262
263impl Default for BoolOr {
264    fn default() -> Self {
265        Self::new()
266    }
267}
268
269impl AggregateUDFImpl for BoolOr {
270    fn as_any(&self) -> &dyn Any {
271        self
272    }
273
274    fn name(&self) -> &str {
275        "bool_or"
276    }
277
278    fn signature(&self) -> &Signature {
279        &self.signature
280    }
281
282    fn return_type(&self, _: &[DataType]) -> Result<DataType> {
283        Ok(DataType::Boolean)
284    }
285
286    fn accumulator(&self, _: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
287        Ok(Box::<BoolOrAccumulator>::default())
288    }
289
290    fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
291        Ok(vec![
292            Field::new(
293                format_state_name(args.name, self.name()),
294                DataType::Boolean,
295                true,
296            )
297            .into(),
298        ])
299    }
300
301    fn groups_accumulator_supported(&self, _args: AccumulatorArgs) -> bool {
302        true
303    }
304
305    fn create_groups_accumulator(
306        &self,
307        args: AccumulatorArgs,
308    ) -> Result<Box<dyn GroupsAccumulator>> {
309        match args.return_field.data_type() {
310            DataType::Boolean => Ok(Box::new(BooleanGroupsAccumulator::new(
311                |x, y| x || y,
312                false,
313            ))),
314            _ => not_impl_err!(
315                "GroupsAccumulator not supported for {} with {}",
316                args.name,
317                args.return_field.data_type()
318            ),
319        }
320    }
321
322    fn order_sensitivity(&self) -> AggregateOrderSensitivity {
323        AggregateOrderSensitivity::Insensitive
324    }
325
326    fn reverse_expr(&self) -> ReversedUDAF {
327        ReversedUDAF::Identical
328    }
329
330    fn documentation(&self) -> Option<&Documentation> {
331        self.doc()
332    }
333}
334
335#[derive(Debug, Default)]
336struct BoolOrAccumulator {
337    acc: Option<bool>,
338}
339
340impl Accumulator for BoolOrAccumulator {
341    fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
342        let values = &values[0];
343        self.acc = match (self.acc, bool_or_batch(values)?) {
344            (None, ScalarValue::Boolean(v)) => v,
345            (Some(v), ScalarValue::Boolean(None)) => Some(v),
346            (Some(a), ScalarValue::Boolean(Some(b))) => Some(a || b),
347            _ => unreachable!(),
348        };
349        Ok(())
350    }
351
352    fn evaluate(&mut self) -> Result<ScalarValue> {
353        Ok(ScalarValue::Boolean(self.acc))
354    }
355
356    fn size(&self) -> usize {
357        size_of_val(self)
358    }
359
360    fn state(&mut self) -> Result<Vec<ScalarValue>> {
361        Ok(vec![ScalarValue::Boolean(self.acc)])
362    }
363
364    fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
365        self.update_batch(states)
366    }
367}
datafusion_functions_aggregate/bool_and_or.rs

datafusion_functions_aggregate/
bool_and_or.rs
