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

datafusion_functions_aggregate/
bool_and_or.rs
