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::DataType;
28use arrow::datatypes::Field;
29
30use datafusion_common::internal_err;
31use datafusion_common::{downcast_value, not_impl_err};
32use datafusion_common::{Result, ScalarValue};
33use datafusion_expr::function::{AccumulatorArgs, StateFieldsArgs};
34use datafusion_expr::utils::{format_state_name, AggregateOrderSensitivity};
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)]
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<Field>> {
154        Ok(vec![Field::new(
155            format_state_name(args.name, self.name()),
156            DataType::Boolean,
157            true,
158        )])
159    }
160
161    fn groups_accumulator_supported(&self, _args: AccumulatorArgs) -> bool {
162        true
163    }
164
165    fn create_groups_accumulator(
166        &self,
167        args: AccumulatorArgs,
168    ) -> Result<Box<dyn GroupsAccumulator>> {
169        match args.return_type {
170            DataType::Boolean => {
171                Ok(Box::new(BooleanGroupsAccumulator::new(|x, y| x && y, true)))
172            }
173            _ => not_impl_err!(
174                "GroupsAccumulator not supported for {} with {}",
175                args.name,
176                args.return_type
177            ),
178        }
179    }
180
181    fn aliases(&self) -> &[String] {
182        &[]
183    }
184
185    fn order_sensitivity(&self) -> AggregateOrderSensitivity {
186        AggregateOrderSensitivity::Insensitive
187    }
188
189    fn reverse_expr(&self) -> ReversedUDAF {
190        ReversedUDAF::Identical
191    }
192
193    fn documentation(&self) -> Option<&Documentation> {
194        self.doc()
195    }
196}
197
198#[derive(Debug, Default)]
199struct BoolAndAccumulator {
200    acc: Option<bool>,
201}
202
203impl Accumulator for BoolAndAccumulator {
204    fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
205        let values = &values[0];
206        self.acc = match (self.acc, bool_and_batch(values)?) {
207            (None, ScalarValue::Boolean(v)) => v,
208            (Some(v), ScalarValue::Boolean(None)) => Some(v),
209            (Some(a), ScalarValue::Boolean(Some(b))) => Some(a && b),
210            _ => unreachable!(),
211        };
212        Ok(())
213    }
214
215    fn evaluate(&mut self) -> Result<ScalarValue> {
216        Ok(ScalarValue::Boolean(self.acc))
217    }
218
219    fn size(&self) -> usize {
220        size_of_val(self)
221    }
222
223    fn state(&mut self) -> Result<Vec<ScalarValue>> {
224        Ok(vec![ScalarValue::Boolean(self.acc)])
225    }
226
227    fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
228        self.update_batch(states)
229    }
230}
231
232#[user_doc(
233    doc_section(label = "General Functions"),
234    description = "Returns true if all non-null input values are true, otherwise false.",
235    syntax_example = "bool_and(expression)",
236    sql_example = r#"```sql
237> SELECT bool_and(column_name) FROM table_name;
238+----------------------------+
239| bool_and(column_name)       |
240+----------------------------+
241| true                        |
242+----------------------------+
243```"#,
244    standard_argument(name = "expression", prefix = "The")
245)]
246/// BOOL_OR aggregate expression
247#[derive(Debug, Clone)]
248pub struct BoolOr {
249    signature: Signature,
250}
251
252impl BoolOr {
253    fn new() -> Self {
254        Self {
255            signature: Signature::uniform(
256                1,
257                vec![DataType::Boolean],
258                Volatility::Immutable,
259            ),
260        }
261    }
262}
263
264impl Default for BoolOr {
265    fn default() -> Self {
266        Self::new()
267    }
268}
269
270impl AggregateUDFImpl for BoolOr {
271    fn as_any(&self) -> &dyn Any {
272        self
273    }
274
275    fn name(&self) -> &str {
276        "bool_or"
277    }
278
279    fn signature(&self) -> &Signature {
280        &self.signature
281    }
282
283    fn return_type(&self, _: &[DataType]) -> Result<DataType> {
284        Ok(DataType::Boolean)
285    }
286
287    fn accumulator(&self, _: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
288        Ok(Box::<BoolOrAccumulator>::default())
289    }
290
291    fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<Field>> {
292        Ok(vec![Field::new(
293            format_state_name(args.name, self.name()),
294            DataType::Boolean,
295            true,
296        )])
297    }
298
299    fn groups_accumulator_supported(&self, _args: AccumulatorArgs) -> bool {
300        true
301    }
302
303    fn create_groups_accumulator(
304        &self,
305        args: AccumulatorArgs,
306    ) -> Result<Box<dyn GroupsAccumulator>> {
307        match args.return_type {
308            DataType::Boolean => Ok(Box::new(BooleanGroupsAccumulator::new(
309                |x, y| x || y,
310                false,
311            ))),
312            _ => not_impl_err!(
313                "GroupsAccumulator not supported for {} with {}",
314                args.name,
315                args.return_type
316            ),
317        }
318    }
319
320    fn aliases(&self) -> &[String] {
321        &[]
322    }
323
324    fn order_sensitivity(&self) -> AggregateOrderSensitivity {
325        AggregateOrderSensitivity::Insensitive
326    }
327
328    fn reverse_expr(&self) -> ReversedUDAF {
329        ReversedUDAF::Identical
330    }
331
332    fn documentation(&self) -> Option<&Documentation> {
333        self.doc()
334    }
335}
336
337#[derive(Debug, Default)]
338struct BoolOrAccumulator {
339    acc: Option<bool>,
340}
341
342impl Accumulator for BoolOrAccumulator {
343    fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
344        let values = &values[0];
345        self.acc = match (self.acc, bool_or_batch(values)?) {
346            (None, ScalarValue::Boolean(v)) => v,
347            (Some(v), ScalarValue::Boolean(None)) => Some(v),
348            (Some(a), ScalarValue::Boolean(Some(b))) => Some(a || b),
349            _ => unreachable!(),
350        };
351        Ok(())
352    }
353
354    fn evaluate(&mut self) -> Result<ScalarValue> {
355        Ok(ScalarValue::Boolean(self.acc))
356    }
357
358    fn size(&self) -> usize {
359        size_of_val(self)
360    }
361
362    fn state(&mut self) -> Result<Vec<ScalarValue>> {
363        Ok(vec![ScalarValue::Boolean(self.acc)])
364    }
365
366    fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
367        self.update_batch(states)
368    }
369}
datafusion_functions_aggregate/bool_and_or.rs

datafusion_functions_aggregate/
bool_and_or.rs
