datafusion_physical_expr/aggregate.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
18pub(crate) mod groups_accumulator {
19 #[allow(unused_imports)]
20 pub(crate) mod accumulate {
21 pub use datafusion_functions_aggregate_common::aggregate::groups_accumulator::accumulate::NullState;
22 }
23 pub use datafusion_functions_aggregate_common::aggregate::groups_accumulator::{
24 accumulate::NullState, GroupsAccumulatorAdapter,
25 };
26}
27pub(crate) mod stats {
28 pub use datafusion_functions_aggregate_common::stats::StatsType;
29}
30pub mod utils {
31 pub use datafusion_functions_aggregate_common::utils::{
32 get_accum_scalar_values_as_arrays, get_sort_options, ordering_fields,
33 DecimalAverager, Hashable,
34 };
35}
36
37use std::fmt::Debug;
38use std::sync::Arc;
39
40use crate::expressions::Column;
41
42use arrow::compute::SortOptions;
43use arrow::datatypes::{DataType, FieldRef, Schema, SchemaRef};
44use datafusion_common::{internal_err, not_impl_err, Result, ScalarValue};
45use datafusion_expr::{AggregateUDF, ReversedUDAF, SetMonotonicity};
46use datafusion_expr_common::accumulator::Accumulator;
47use datafusion_expr_common::groups_accumulator::GroupsAccumulator;
48use datafusion_expr_common::type_coercion::aggregates::check_arg_count;
49use datafusion_functions_aggregate_common::accumulator::{
50 AccumulatorArgs, StateFieldsArgs,
51};
52use datafusion_functions_aggregate_common::order::AggregateOrderSensitivity;
53use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
54use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
55
56/// Builder for physical [`AggregateFunctionExpr`]
57///
58/// `AggregateFunctionExpr` contains the information necessary to call
59/// an aggregate expression.
60#[derive(Debug, Clone)]
61pub struct AggregateExprBuilder {
62 fun: Arc<AggregateUDF>,
63 /// Physical expressions of the aggregate function
64 args: Vec<Arc<dyn PhysicalExpr>>,
65 alias: Option<String>,
66 /// A human readable name
67 human_display: String,
68 /// Arrow Schema for the aggregate function
69 schema: SchemaRef,
70 /// The physical order by expressions
71 order_bys: Vec<PhysicalSortExpr>,
72 /// Whether to ignore null values
73 ignore_nulls: bool,
74 /// Whether is distinct aggregate function
75 is_distinct: bool,
76 /// Whether the expression is reversed
77 is_reversed: bool,
78}
79
80impl AggregateExprBuilder {
81 pub fn new(fun: Arc<AggregateUDF>, args: Vec<Arc<dyn PhysicalExpr>>) -> Self {
82 Self {
83 fun,
84 args,
85 alias: None,
86 human_display: String::default(),
87 schema: Arc::new(Schema::empty()),
88 order_bys: vec![],
89 ignore_nulls: false,
90 is_distinct: false,
91 is_reversed: false,
92 }
93 }
94
95 /// Constructs an `AggregateFunctionExpr` from the builder
96 ///
97 /// Note that an [`Self::alias`] must be provided before calling this method.
98 ///
99 /// # Example: Create an [`AggregateUDF`]
100 ///
101 /// In the following example, [`AggregateFunctionExpr`] will be built using [`AggregateExprBuilder`]
102 /// which provides a build function. Full example could be accessed from the source file.
103 ///
104 /// ```
105 /// # use std::any::Any;
106 /// # use std::sync::Arc;
107 /// # use arrow::datatypes::{DataType, FieldRef};
108 /// # use datafusion_common::{Result, ScalarValue};
109 /// # use datafusion_expr::{col, ColumnarValue, Documentation, Signature, Volatility, Expr};
110 /// # use datafusion_expr::{AggregateUDFImpl, AggregateUDF, Accumulator, function::{AccumulatorArgs, StateFieldsArgs}};
111 /// # use arrow::datatypes::Field;
112 /// #
113 /// # #[derive(Debug, Clone, PartialEq, Eq, Hash)]
114 /// # struct FirstValueUdf {
115 /// # signature: Signature,
116 /// # }
117 /// #
118 /// # impl FirstValueUdf {
119 /// # fn new() -> Self {
120 /// # Self {
121 /// # signature: Signature::any(1, Volatility::Immutable),
122 /// # }
123 /// # }
124 /// # }
125 /// #
126 /// # impl AggregateUDFImpl for FirstValueUdf {
127 /// # fn as_any(&self) -> &dyn Any {
128 /// # unimplemented!()
129 /// # }
130 /// #
131 /// # fn name(&self) -> &str {
132 /// # unimplemented!()
133 /// # }
134 /// #
135 /// # fn signature(&self) -> &Signature {
136 /// # unimplemented!()
137 /// # }
138 /// #
139 /// # fn return_type(&self, args: &[DataType]) -> Result<DataType> {
140 /// # unimplemented!()
141 /// # }
142 /// #
143 /// # fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
144 /// # unimplemented!()
145 /// # }
146 /// #
147 /// # fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
148 /// # unimplemented!()
149 /// # }
150 /// #
151 /// # fn documentation(&self) -> Option<&Documentation> {
152 /// # unimplemented!()
153 /// # }
154 /// # }
155 /// #
156 /// # let first_value = AggregateUDF::from(FirstValueUdf::new());
157 /// # let expr = first_value.call(vec![col("a")]);
158 /// #
159 /// # use datafusion_physical_expr::expressions::Column;
160 /// # use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
161 /// # use datafusion_physical_expr::aggregate::AggregateExprBuilder;
162 /// # use datafusion_physical_expr::expressions::PhysicalSortExpr;
163 /// # use datafusion_physical_expr::PhysicalSortRequirement;
164 /// #
165 /// fn build_aggregate_expr() -> Result<()> {
166 /// let args = vec![Arc::new(Column::new("a", 0)) as Arc<dyn PhysicalExpr>];
167 /// let order_by = vec![PhysicalSortExpr {
168 /// expr: Arc::new(Column::new("x", 1)) as Arc<dyn PhysicalExpr>,
169 /// options: Default::default(),
170 /// }];
171 ///
172 /// let first_value = AggregateUDF::from(FirstValueUdf::new());
173 ///
174 /// let aggregate_expr = AggregateExprBuilder::new(
175 /// Arc::new(first_value),
176 /// args
177 /// )
178 /// .order_by(order_by)
179 /// .alias("first_a_by_x")
180 /// .ignore_nulls()
181 /// .build()?;
182 ///
183 /// Ok(())
184 /// }
185 /// ```
186 ///
187 /// This creates a physical expression equivalent to SQL:
188 /// `first_value(a ORDER BY x) IGNORE NULLS AS first_a_by_x`
189 pub fn build(self) -> Result<AggregateFunctionExpr> {
190 let Self {
191 fun,
192 args,
193 alias,
194 human_display,
195 schema,
196 order_bys,
197 ignore_nulls,
198 is_distinct,
199 is_reversed,
200 } = self;
201 if args.is_empty() {
202 return internal_err!("args should not be empty");
203 }
204
205 let ordering_types = order_bys
206 .iter()
207 .map(|e| e.expr.data_type(&schema))
208 .collect::<Result<Vec<_>>>()?;
209
210 let ordering_fields = utils::ordering_fields(&order_bys, &ordering_types);
211
212 let input_exprs_fields = args
213 .iter()
214 .map(|arg| arg.return_field(&schema))
215 .collect::<Result<Vec<_>>>()?;
216
217 check_arg_count(
218 fun.name(),
219 &input_exprs_fields,
220 &fun.signature().type_signature,
221 )?;
222
223 let return_field = fun.return_field(&input_exprs_fields)?;
224 let is_nullable = fun.is_nullable();
225 let name = match alias {
226 None => {
227 return internal_err!(
228 "AggregateExprBuilder::alias must be provided prior to calling build"
229 )
230 }
231 Some(alias) => alias,
232 };
233
234 Ok(AggregateFunctionExpr {
235 fun: Arc::unwrap_or_clone(fun),
236 args,
237 return_field,
238 name,
239 human_display,
240 schema: Arc::unwrap_or_clone(schema),
241 order_bys,
242 ignore_nulls,
243 ordering_fields,
244 is_distinct,
245 input_fields: input_exprs_fields,
246 is_reversed,
247 is_nullable,
248 })
249 }
250
251 pub fn alias(mut self, alias: impl Into<String>) -> Self {
252 self.alias = Some(alias.into());
253 self
254 }
255
256 pub fn human_display(mut self, name: String) -> Self {
257 self.human_display = name;
258 self
259 }
260
261 pub fn schema(mut self, schema: SchemaRef) -> Self {
262 self.schema = schema;
263 self
264 }
265
266 pub fn order_by(mut self, order_bys: Vec<PhysicalSortExpr>) -> Self {
267 self.order_bys = order_bys;
268 self
269 }
270
271 pub fn reversed(mut self) -> Self {
272 self.is_reversed = true;
273 self
274 }
275
276 pub fn with_reversed(mut self, is_reversed: bool) -> Self {
277 self.is_reversed = is_reversed;
278 self
279 }
280
281 pub fn distinct(mut self) -> Self {
282 self.is_distinct = true;
283 self
284 }
285
286 pub fn with_distinct(mut self, is_distinct: bool) -> Self {
287 self.is_distinct = is_distinct;
288 self
289 }
290
291 pub fn ignore_nulls(mut self) -> Self {
292 self.ignore_nulls = true;
293 self
294 }
295
296 pub fn with_ignore_nulls(mut self, ignore_nulls: bool) -> Self {
297 self.ignore_nulls = ignore_nulls;
298 self
299 }
300}
301
302/// Physical aggregate expression of a UDAF.
303///
304/// Instances are constructed via [`AggregateExprBuilder`].
305#[derive(Debug, Clone)]
306pub struct AggregateFunctionExpr {
307 fun: AggregateUDF,
308 args: Vec<Arc<dyn PhysicalExpr>>,
309 /// Output / return field of this aggregate
310 return_field: FieldRef,
311 /// Output column name that this expression creates
312 name: String,
313 /// Simplified name for `tree` explain.
314 human_display: String,
315 schema: Schema,
316 // The physical order by expressions
317 order_bys: Vec<PhysicalSortExpr>,
318 // Whether to ignore null values
319 ignore_nulls: bool,
320 // fields used for order sensitive aggregation functions
321 ordering_fields: Vec<FieldRef>,
322 is_distinct: bool,
323 is_reversed: bool,
324 input_fields: Vec<FieldRef>,
325 is_nullable: bool,
326}
327
328impl AggregateFunctionExpr {
329 /// Return the `AggregateUDF` used by this `AggregateFunctionExpr`
330 pub fn fun(&self) -> &AggregateUDF {
331 &self.fun
332 }
333
334 /// expressions that are passed to the Accumulator.
335 /// Single-column aggregations such as `sum` return a single value, others (e.g. `cov`) return many.
336 pub fn expressions(&self) -> Vec<Arc<dyn PhysicalExpr>> {
337 self.args.clone()
338 }
339
340 /// Human readable name such as `"MIN(c2)"`.
341 pub fn name(&self) -> &str {
342 &self.name
343 }
344
345 /// Simplified name for `tree` explain.
346 pub fn human_display(&self) -> &str {
347 &self.human_display
348 }
349
350 /// Return if the aggregation is distinct
351 pub fn is_distinct(&self) -> bool {
352 self.is_distinct
353 }
354
355 /// Return if the aggregation ignores nulls
356 pub fn ignore_nulls(&self) -> bool {
357 self.ignore_nulls
358 }
359
360 /// Return if the aggregation is reversed
361 pub fn is_reversed(&self) -> bool {
362 self.is_reversed
363 }
364
365 /// Return if the aggregation is nullable
366 pub fn is_nullable(&self) -> bool {
367 self.is_nullable
368 }
369
370 /// the field of the final result of this aggregation.
371 pub fn field(&self) -> FieldRef {
372 self.return_field
373 .as_ref()
374 .clone()
375 .with_name(&self.name)
376 .into()
377 }
378
379 /// the accumulator used to accumulate values from the expressions.
380 /// the accumulator expects the same number of arguments as `expressions` and must
381 /// return states with the same description as `state_fields`
382 pub fn create_accumulator(&self) -> Result<Box<dyn Accumulator>> {
383 let acc_args = AccumulatorArgs {
384 return_field: Arc::clone(&self.return_field),
385 schema: &self.schema,
386 ignore_nulls: self.ignore_nulls,
387 order_bys: self.order_bys.as_ref(),
388 is_distinct: self.is_distinct,
389 name: &self.name,
390 is_reversed: self.is_reversed,
391 exprs: &self.args,
392 };
393
394 self.fun.accumulator(acc_args)
395 }
396
397 /// the field of the final result of this aggregation.
398 pub fn state_fields(&self) -> Result<Vec<FieldRef>> {
399 let args = StateFieldsArgs {
400 name: &self.name,
401 input_fields: &self.input_fields,
402 return_field: Arc::clone(&self.return_field),
403 ordering_fields: &self.ordering_fields,
404 is_distinct: self.is_distinct,
405 };
406
407 self.fun.state_fields(args)
408 }
409
410 /// Returns the ORDER BY expressions for the aggregate function.
411 pub fn order_bys(&self) -> &[PhysicalSortExpr] {
412 if self.order_sensitivity().is_insensitive() {
413 &[]
414 } else {
415 &self.order_bys
416 }
417 }
418
419 /// Indicates whether aggregator can produce the correct result with any
420 /// arbitrary input ordering. By default, we assume that aggregate expressions
421 /// are order insensitive.
422 pub fn order_sensitivity(&self) -> AggregateOrderSensitivity {
423 if self.order_bys.is_empty() {
424 AggregateOrderSensitivity::Insensitive
425 } else {
426 // If there is an ORDER BY clause, use the sensitivity of the implementation:
427 self.fun.order_sensitivity()
428 }
429 }
430
431 /// Sets the indicator whether ordering requirements of the aggregator is
432 /// satisfied by its input. If this is not the case, aggregators with order
433 /// sensitivity `AggregateOrderSensitivity::Beneficial` can still produce
434 /// the correct result with possibly more work internally.
435 ///
436 /// # Returns
437 ///
438 /// Returns `Ok(Some(updated_expr))` if the process completes successfully.
439 /// If the expression can benefit from existing input ordering, but does
440 /// not implement the method, returns an error. Order insensitive and hard
441 /// requirement aggregators return `Ok(None)`.
442 pub fn with_beneficial_ordering(
443 self: Arc<Self>,
444 beneficial_ordering: bool,
445 ) -> Result<Option<AggregateFunctionExpr>> {
446 let Some(updated_fn) = self
447 .fun
448 .clone()
449 .with_beneficial_ordering(beneficial_ordering)?
450 else {
451 return Ok(None);
452 };
453
454 AggregateExprBuilder::new(Arc::new(updated_fn), self.args.to_vec())
455 .order_by(self.order_bys.clone())
456 .schema(Arc::new(self.schema.clone()))
457 .alias(self.name().to_string())
458 .with_ignore_nulls(self.ignore_nulls)
459 .with_distinct(self.is_distinct)
460 .with_reversed(self.is_reversed)
461 .build()
462 .map(Some)
463 }
464
465 /// Creates accumulator implementation that supports retract
466 pub fn create_sliding_accumulator(&self) -> Result<Box<dyn Accumulator>> {
467 let args = AccumulatorArgs {
468 return_field: Arc::clone(&self.return_field),
469 schema: &self.schema,
470 ignore_nulls: self.ignore_nulls,
471 order_bys: self.order_bys.as_ref(),
472 is_distinct: self.is_distinct,
473 name: &self.name,
474 is_reversed: self.is_reversed,
475 exprs: &self.args,
476 };
477
478 let accumulator = self.fun.create_sliding_accumulator(args)?;
479
480 // Accumulators that have window frame startings different
481 // than `UNBOUNDED PRECEDING`, such as `1 PRECEDING`, need to
482 // implement retract_batch method in order to run correctly
483 // currently in DataFusion.
484 //
485 // If this `retract_batches` is not present, there is no way
486 // to calculate result correctly. For example, the query
487 //
488 // ```sql
489 // SELECT
490 // SUM(a) OVER(ORDER BY a ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING) AS sum_a
491 // FROM
492 // t
493 // ```
494 //
495 // 1. First sum value will be the sum of rows between `[0, 1)`,
496 //
497 // 2. Second sum value will be the sum of rows between `[0, 2)`
498 //
499 // 3. Third sum value will be the sum of rows between `[1, 3)`, etc.
500 //
501 // Since the accumulator keeps the running sum:
502 //
503 // 1. First sum we add to the state sum value between `[0, 1)`
504 //
505 // 2. Second sum we add to the state sum value between `[1, 2)`
506 // (`[0, 1)` is already in the state sum, hence running sum will
507 // cover `[0, 2)` range)
508 //
509 // 3. Third sum we add to the state sum value between `[2, 3)`
510 // (`[0, 2)` is already in the state sum). Also we need to
511 // retract values between `[0, 1)` by this way we can obtain sum
512 // between [1, 3) which is indeed the appropriate range.
513 //
514 // When we use `UNBOUNDED PRECEDING` in the query starting
515 // index will always be 0 for the desired range, and hence the
516 // `retract_batch` method will not be called. In this case
517 // having retract_batch is not a requirement.
518 //
519 // This approach is a a bit different than window function
520 // approach. In window function (when they use a window frame)
521 // they get all the desired range during evaluation.
522 if !accumulator.supports_retract_batch() {
523 return not_impl_err!(
524 "Aggregate can not be used as a sliding accumulator because \
525 `retract_batch` is not implemented: {}",
526 self.name
527 );
528 }
529 Ok(accumulator)
530 }
531
532 /// If the aggregate expression has a specialized
533 /// [`GroupsAccumulator`] implementation. If this returns true,
534 /// `[Self::create_groups_accumulator`] will be called.
535 pub fn groups_accumulator_supported(&self) -> bool {
536 let args = AccumulatorArgs {
537 return_field: Arc::clone(&self.return_field),
538 schema: &self.schema,
539 ignore_nulls: self.ignore_nulls,
540 order_bys: self.order_bys.as_ref(),
541 is_distinct: self.is_distinct,
542 name: &self.name,
543 is_reversed: self.is_reversed,
544 exprs: &self.args,
545 };
546 self.fun.groups_accumulator_supported(args)
547 }
548
549 /// Return a specialized [`GroupsAccumulator`] that manages state
550 /// for all groups.
551 ///
552 /// For maximum performance, a [`GroupsAccumulator`] should be
553 /// implemented in addition to [`Accumulator`].
554 pub fn create_groups_accumulator(&self) -> Result<Box<dyn GroupsAccumulator>> {
555 let args = AccumulatorArgs {
556 return_field: Arc::clone(&self.return_field),
557 schema: &self.schema,
558 ignore_nulls: self.ignore_nulls,
559 order_bys: self.order_bys.as_ref(),
560 is_distinct: self.is_distinct,
561 name: &self.name,
562 is_reversed: self.is_reversed,
563 exprs: &self.args,
564 };
565 self.fun.create_groups_accumulator(args)
566 }
567
568 /// Construct an expression that calculates the aggregate in reverse.
569 /// Typically the "reverse" expression is itself (e.g. SUM, COUNT).
570 /// For aggregates that do not support calculation in reverse,
571 /// returns None (which is the default value).
572 pub fn reverse_expr(&self) -> Option<AggregateFunctionExpr> {
573 match self.fun.reverse_udf() {
574 ReversedUDAF::NotSupported => None,
575 ReversedUDAF::Identical => Some(self.clone()),
576 ReversedUDAF::Reversed(reverse_udf) => {
577 let mut name = self.name().to_string();
578 // If the function is changed, we need to reverse order_by clause as well
579 // i.e. First(a order by b asc null first) -> Last(a order by b desc null last)
580 if self.fun().name() != reverse_udf.name() {
581 replace_order_by_clause(&mut name);
582 }
583 replace_fn_name_clause(&mut name, self.fun.name(), reverse_udf.name());
584
585 AggregateExprBuilder::new(reverse_udf, self.args.to_vec())
586 .order_by(self.order_bys.iter().map(|e| e.reverse()).collect())
587 .schema(Arc::new(self.schema.clone()))
588 .alias(name)
589 .with_ignore_nulls(self.ignore_nulls)
590 .with_distinct(self.is_distinct)
591 .with_reversed(!self.is_reversed)
592 .build()
593 .ok()
594 }
595 }
596 }
597
598 /// Returns all expressions used in the [`AggregateFunctionExpr`].
599 /// These expressions are (1)function arguments, (2) order by expressions.
600 pub fn all_expressions(&self) -> AggregatePhysicalExpressions {
601 let args = self.expressions();
602 let order_by_exprs = self
603 .order_bys()
604 .iter()
605 .map(|sort_expr| Arc::clone(&sort_expr.expr))
606 .collect();
607 AggregatePhysicalExpressions {
608 args,
609 order_by_exprs,
610 }
611 }
612
613 /// Rewrites [`AggregateFunctionExpr`], with new expressions given. The argument should be consistent
614 /// with the return value of the [`AggregateFunctionExpr::all_expressions`] method.
615 /// Returns `Some(Arc<dyn AggregateExpr>)` if re-write is supported, otherwise returns `None`.
616 pub fn with_new_expressions(
617 &self,
618 args: Vec<Arc<dyn PhysicalExpr>>,
619 order_by_exprs: Vec<Arc<dyn PhysicalExpr>>,
620 ) -> Option<AggregateFunctionExpr> {
621 if args.len() != self.args.len()
622 || (self.order_sensitivity() != AggregateOrderSensitivity::Insensitive
623 && order_by_exprs.len() != self.order_bys.len())
624 {
625 return None;
626 }
627
628 let new_order_bys = self
629 .order_bys
630 .iter()
631 .zip(order_by_exprs)
632 .map(|(req, new_expr)| PhysicalSortExpr {
633 expr: new_expr,
634 options: req.options,
635 })
636 .collect();
637
638 Some(AggregateFunctionExpr {
639 fun: self.fun.clone(),
640 args,
641 return_field: Arc::clone(&self.return_field),
642 name: self.name.clone(),
643 // TODO: Human name should be updated after re-write to not mislead
644 human_display: self.human_display.clone(),
645 schema: self.schema.clone(),
646 order_bys: new_order_bys,
647 ignore_nulls: self.ignore_nulls,
648 ordering_fields: self.ordering_fields.clone(),
649 is_distinct: self.is_distinct,
650 is_reversed: false,
651 input_fields: self.input_fields.clone(),
652 is_nullable: self.is_nullable,
653 })
654 }
655
656 /// If this function is max, return (output_field, true)
657 /// if the function is min, return (output_field, false)
658 /// otherwise return None (the default)
659 ///
660 /// output_field is the name of the column produced by this aggregate
661 ///
662 /// Note: this is used to use special aggregate implementations in certain conditions
663 pub fn get_minmax_desc(&self) -> Option<(FieldRef, bool)> {
664 self.fun.is_descending().map(|flag| (self.field(), flag))
665 }
666
667 /// Returns default value of the function given the input is Null
668 /// Most of the aggregate function return Null if input is Null,
669 /// while `count` returns 0 if input is Null
670 pub fn default_value(&self, data_type: &DataType) -> Result<ScalarValue> {
671 self.fun.default_value(data_type)
672 }
673
674 /// Indicates whether the aggregation function is monotonic as a set
675 /// function. See [`SetMonotonicity`] for details.
676 pub fn set_monotonicity(&self) -> SetMonotonicity {
677 let field = self.field();
678 let data_type = field.data_type();
679 self.fun.inner().set_monotonicity(data_type)
680 }
681
682 /// Returns `PhysicalSortExpr` based on the set monotonicity of the function.
683 pub fn get_result_ordering(&self, aggr_func_idx: usize) -> Option<PhysicalSortExpr> {
684 // If the aggregate expressions are set-monotonic, the output data is
685 // naturally ordered with it per group or partition.
686 let monotonicity = self.set_monotonicity();
687 if monotonicity == SetMonotonicity::NotMonotonic {
688 return None;
689 }
690 let expr = Arc::new(Column::new(self.name(), aggr_func_idx));
691 let options =
692 SortOptions::new(monotonicity == SetMonotonicity::Decreasing, false);
693 Some(PhysicalSortExpr { expr, options })
694 }
695}
696
697/// Stores the physical expressions used inside the `AggregateExpr`.
698pub struct AggregatePhysicalExpressions {
699 /// Aggregate function arguments
700 pub args: Vec<Arc<dyn PhysicalExpr>>,
701 /// Order by expressions
702 pub order_by_exprs: Vec<Arc<dyn PhysicalExpr>>,
703}
704
705impl PartialEq for AggregateFunctionExpr {
706 fn eq(&self, other: &Self) -> bool {
707 self.name == other.name
708 && self.return_field == other.return_field
709 && self.fun == other.fun
710 && self.args.len() == other.args.len()
711 && self
712 .args
713 .iter()
714 .zip(other.args.iter())
715 .all(|(this_arg, other_arg)| this_arg.eq(other_arg))
716 }
717}
718
719fn replace_order_by_clause(order_by: &mut String) {
720 let suffixes = [
721 (" DESC NULLS FIRST]", " ASC NULLS LAST]"),
722 (" ASC NULLS FIRST]", " DESC NULLS LAST]"),
723 (" DESC NULLS LAST]", " ASC NULLS FIRST]"),
724 (" ASC NULLS LAST]", " DESC NULLS FIRST]"),
725 ];
726
727 if let Some(start) = order_by.find("ORDER BY [") {
728 if let Some(end) = order_by[start..].find(']') {
729 let order_by_start = start + 9;
730 let order_by_end = start + end;
731
732 let column_order = &order_by[order_by_start..=order_by_end];
733 for (suffix, replacement) in suffixes {
734 if column_order.ends_with(suffix) {
735 let new_order = column_order.replace(suffix, replacement);
736 order_by.replace_range(order_by_start..=order_by_end, &new_order);
737 break;
738 }
739 }
740 }
741 }
742}
743
744fn replace_fn_name_clause(aggr_name: &mut String, fn_name_old: &str, fn_name_new: &str) {
745 *aggr_name = aggr_name.replace(fn_name_old, fn_name_new);
746}